From d805bd66ed8de28a3e98c990a4b9c2093018b791 Mon Sep 17 00:00:00 2001 From: Mateus Azis Date: Thu, 26 Oct 2023 11:09:44 -0700 Subject: [PATCH 01/11] Handle slow security policies without blocking gRPC threads. - Introduce PendingAuthListener to handle a ListenableFuture, progressing the gRPC through each stage in sequence once the future completes and is OK. - Move unit tests away from `checkAuthorizationForService` and into `checkAuthorizationForServiceAsync` since that should be the only method called in production now. - Some tests in `ServerSecurityPolicyTest` had their expectations updated; they previously called synchornous APIs that transformed failed `ListenableFuture` into one or another status. Now, we call the sync API, so those transformations do not happen anymore, thus the test needs to deal with failed futures directly. - I couldn't figure out if this PR needs extra tests. AFAICT `BinderSecurityTest` should already cover the new codepaths, but please let me know otherwise. This should be the last PR to address #10566. --- .../io/grpc/binder/BinderServerBuilder.java | 2 +- .../internal/BinderTransportSecurity.java | 53 +++---- .../binder/internal/PendingAuthListener.java | 97 +++++++++++++ .../grpc/binder/ServerSecurityPolicyTest.java | 134 ++++++++++-------- 4 files changed, 194 insertions(+), 92 deletions(-) create mode 100644 binder/src/main/java/io/grpc/binder/internal/PendingAuthListener.java diff --git a/binder/src/main/java/io/grpc/binder/BinderServerBuilder.java b/binder/src/main/java/io/grpc/binder/BinderServerBuilder.java index dafb884d6b1..fe3d71ebf1d 100644 --- a/binder/src/main/java/io/grpc/binder/BinderServerBuilder.java +++ b/binder/src/main/java/io/grpc/binder/BinderServerBuilder.java @@ -171,7 +171,7 @@ public Server build() { checkState(!isBuilt, "BinderServerBuilder can only be used to build one server instance."); isBuilt = true; // We install the security interceptor last, so it's closest to the transport. - BinderTransportSecurity.installAuthInterceptor(this); + BinderTransportSecurity.installAuthInterceptor(this, serverImplBuilder.getExecutorPool()); return super.build(); } } 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 022ae7f6d20..5eacca49192 100644 --- a/binder/src/main/java/io/grpc/binder/internal/BinderTransportSecurity.java +++ b/binder/src/main/java/io/grpc/binder/internal/BinderTransportSecurity.java @@ -28,9 +28,11 @@ import io.grpc.ServerInterceptor; import io.grpc.Status; import io.grpc.internal.GrpcAttributes; +import io.grpc.internal.ObjectPool; import java.util.concurrent.ConcurrentHashMap; -import java.util.concurrent.ExecutionException; +import java.util.concurrent.Executor; import javax.annotation.CheckReturnValue; +import javax.annotation.Nullable; /** * Manages security for an Android Service hosted gRPC server. @@ -51,8 +53,8 @@ private BinderTransportSecurity() {} * @param serverBuilder The ServerBuilder being used to create the server. */ @Internal - public static void installAuthInterceptor(ServerBuilder serverBuilder) { - serverBuilder.intercept(new ServerAuthInterceptor()); + public static void installAuthInterceptor(ServerBuilder serverBuilder, ObjectPool executorPool) { + serverBuilder.intercept(new ServerAuthInterceptor(executorPool)); } /** @@ -78,30 +80,33 @@ public static void attachAuthAttrs( * Authentication state is fetched from the call attributes, inherited from the transport. */ private static final class ServerAuthInterceptor implements ServerInterceptor { + + private final ObjectPool executorPool; + + ServerAuthInterceptor(ObjectPool executorPool) { + this.executorPool = executorPool; + } + @Override public ServerCall.Listener interceptCall( ServerCall call, Metadata headers, ServerCallHandler next) { - Status authStatus = + ListenableFuture authStatusFuture = call.getAttributes() .get(TRANSPORT_AUTHORIZATION_STATE) .checkAuthorization(call.getMethodDescriptor()); - if (authStatus.isOk()) { - return next.startCall(call, headers); - } else { - call.close(authStatus, new Metadata()); - return new ServerCall.Listener() {}; - } + + return new PendingAuthListener<>(authStatusFuture, executorPool, call, headers, next); } } /** - * Maintaines the authorization state for a single transport instance. This class lives for the + * Maintains the authorization state for a single transport instance. This class lives for the * lifetime of a single transport. */ private static final class TransportAuthorizationState { private final int uid; private final ServerPolicyChecker serverPolicyChecker; - private final ConcurrentHashMap serviceAuthorization; + private final ConcurrentHashMap> serviceAuthorization; TransportAuthorizationState(int uid, ServerPolicyChecker serverPolicyChecker) { this.uid = uid; @@ -111,32 +116,20 @@ private static final class TransportAuthorizationState { /** Get whether we're authorized to make this call. */ @CheckReturnValue - Status checkAuthorization(MethodDescriptor method) { + 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, programatically - // created methods could casue this cahe to grow unbounded. + // which is true for all generated methods. Otherwise, programmatically + // created methods could case this cache to grow unbounded. boolean useCache = method.isSampledToLocalTracing(); - Status authorization; if (useCache) { - authorization = serviceAuthorization.get(serviceName); + @Nullable ListenableFuture authorization = serviceAuthorization.get(serviceName); if (authorization != null) { return authorization; } } - try { - // TODO(10566): provide a synchronous version of "checkAuthorization" to avoid blocking the - // calling thread on the completion of the future. - authorization = - serverPolicyChecker.checkAuthorizationForServiceAsync(uid, serviceName).get(); - } catch (ExecutionException e) { - // Do not cache this failure since it may be transient. - return Status.fromThrowable(e); - } catch (InterruptedException e) { - // Do not cache this failure since it may be transient. - Thread.currentThread().interrupt(); - return Status.CANCELLED.withCause(e); - } + ListenableFuture authorization = + serverPolicyChecker.checkAuthorizationForServiceAsync(uid, serviceName); if (useCache) { serviceAuthorization.putIfAbsent(serviceName, authorization); } diff --git a/binder/src/main/java/io/grpc/binder/internal/PendingAuthListener.java b/binder/src/main/java/io/grpc/binder/internal/PendingAuthListener.java new file mode 100644 index 00000000000..13c58c9db01 --- /dev/null +++ b/binder/src/main/java/io/grpc/binder/internal/PendingAuthListener.java @@ -0,0 +1,97 @@ +package io.grpc.binder.internal; + +import com.google.common.util.concurrent.Futures; +import com.google.common.util.concurrent.ListenableFuture; +import com.google.common.util.concurrent.MoreExecutors; + +import io.grpc.Metadata; +import io.grpc.ServerCall; +import io.grpc.ServerCall.Listener; +import io.grpc.ServerCallHandler; +import io.grpc.Status; +import io.grpc.internal.ObjectPool; + +import java.util.concurrent.Executor; + +/** + * A {@link ServerCall.Listener} that can be returned by a {@link io.grpc.ServerInterceptor} to + * asynchronously advance the gRPC pending resolving a possibly asynchronous security policy check. + */ +final class PendingAuthListener extends ServerCall.Listener { + + private final ListenableFuture> authStatusFuture; + private final Executor sequentialExecutor; + private final ObjectPool executorPool; + private final Executor executor; + + /** + * @param authStatusFuture a ListenableFuture holding the result status of the authorization + * policy from a {@link io.grpc.binder.SecurityPolicy} or a + * {@link io.grpc.binder.AsyncSecurityPolicy}. The call only progresses + * if {@link Status#isOk()} is true. + * @param executorPool a pool that can provide at least one Executor under which the result + * of {@code authStatusFuture} can be handled, progressing the gRPC + * stages. + * @param call the 'call' parameter from {@link io.grpc.ServerInterceptor} + * @param headers the 'headers' parameter from {@link io.grpc.ServerInterceptor} + * @param next the 'next' parameter from {@link io.grpc.ServerInterceptor} + */ + PendingAuthListener( + ListenableFuture authStatusFuture, + ObjectPool executorPool, + ServerCall call, Metadata headers, + ServerCallHandler next) { + this.executorPool = executorPool; + this.executor = executorPool.getObject(); + this.authStatusFuture = Futures.transform(authStatusFuture, authStatus -> { + if (authStatus.isOk()) { + return next.startCall(call, headers); + } + call.close(authStatus, new Metadata()); + throw new IllegalStateException("Auth failed", authStatus.asException()); + }, executor); + this.sequentialExecutor = MoreExecutors.newSequentialExecutor(executor); + } + + @Override + public void onCancel() { + ListenableFuture unused = Futures.transform(authStatusFuture, delegate -> { + delegate.onCancel(); + executorPool.returnObject(executor); + return null; + }, sequentialExecutor); + } + + @Override + public void onComplete() { + ListenableFuture unused = Futures.transform(authStatusFuture, delegate -> { + delegate.onComplete(); + executorPool.returnObject(executor); + return null; + }, sequentialExecutor); + } + + @Override + public void onHalfClose() { + ListenableFuture unused = Futures.transform(authStatusFuture, delegate -> { + delegate.onHalfClose(); + return null; + }, sequentialExecutor); + } + + @Override + public void onMessage(ReqT message) { + ListenableFuture unused = Futures.transform(authStatusFuture, delegate -> { + delegate.onMessage(message); + return null; + }, sequentialExecutor); + } + + @Override + public void onReady() { + ListenableFuture unused = Futures.transform(authStatusFuture, delegate -> { + delegate.onReady(); + return null; + }, sequentialExecutor); + } +} diff --git a/binder/src/test/java/io/grpc/binder/ServerSecurityPolicyTest.java b/binder/src/test/java/io/grpc/binder/ServerSecurityPolicyTest.java index 9393aab1e7f..2e663075c35 100644 --- a/binder/src/test/java/io/grpc/binder/ServerSecurityPolicyTest.java +++ b/binder/src/test/java/io/grpc/binder/ServerSecurityPolicyTest.java @@ -17,6 +17,7 @@ package io.grpc.binder; import static com.google.common.truth.Truth.assertThat; +import static org.junit.Assert.assertThrows; import static org.junit.Assert.fail; import android.os.Process; @@ -25,14 +26,18 @@ import com.google.common.util.concurrent.ListenableFuture; import com.google.common.util.concurrent.ListeningExecutorService; import com.google.common.util.concurrent.MoreExecutors; +import com.google.common.util.concurrent.Uninterruptibles; + import io.grpc.Status; -import io.grpc.StatusException; +import io.grpc.Status.Code; import org.junit.Test; import org.junit.runner.RunWith; import org.robolectric.RobolectricTestRunner; import java.util.concurrent.BrokenBarrierException; +import java.util.concurrent.CancellationException; import java.util.concurrent.CountDownLatch; import java.util.concurrent.CyclicBarrier; +import java.util.concurrent.ExecutionException; import java.util.concurrent.Executors; @RunWith(RobolectricTestRunner.class) @@ -48,51 +53,41 @@ public final class ServerSecurityPolicyTest { ServerSecurityPolicy policy; @Test - public void testDefaultInternalOnly() { + public void testDefaultInternalOnly() throws Exception { policy = new ServerSecurityPolicy(); - assertThat(policy.checkAuthorizationForService(MY_UID, SERVICE1).getCode()) - .isEqualTo(Status.OK.getCode()); - assertThat(policy.checkAuthorizationForService(MY_UID, SERVICE2).getCode()) - .isEqualTo(Status.OK.getCode()); + checkSynchronousPolicy(policy, MY_UID, SERVICE1, Code.OK); + checkSynchronousPolicy(policy, MY_UID, SERVICE2, Code.OK); } @Test - public void testInternalOnly_AnotherUid() { + public void testInternalOnly_AnotherUid() throws Exception { policy = new ServerSecurityPolicy(); - assertThat(policy.checkAuthorizationForService(OTHER_UID, SERVICE1).getCode()) - .isEqualTo(Status.PERMISSION_DENIED.getCode()); - assertThat(policy.checkAuthorizationForService(OTHER_UID, SERVICE2).getCode()) - .isEqualTo(Status.PERMISSION_DENIED.getCode()); + checkSynchronousPolicy(policy, OTHER_UID, SERVICE1, Code.PERMISSION_DENIED); + checkSynchronousPolicy(policy, OTHER_UID, SERVICE2, Code.PERMISSION_DENIED); } @Test - public void testBuilderDefault() { + public void testBuilderDefault() throws Exception { policy = ServerSecurityPolicy.newBuilder().build(); - assertThat(policy.checkAuthorizationForService(MY_UID, SERVICE1).getCode()) - .isEqualTo(Status.OK.getCode()); - assertThat(policy.checkAuthorizationForService(OTHER_UID, SERVICE1).getCode()) - .isEqualTo(Status.PERMISSION_DENIED.getCode()); + checkSynchronousPolicy(policy, MY_UID, SERVICE1, Code.OK); + checkSynchronousPolicy(policy, OTHER_UID, SERVICE1, Code.PERMISSION_DENIED); } @Test - public void testPerService() { + public void testPerService() throws Exception { policy = ServerSecurityPolicy.newBuilder() .servicePolicy(SERVICE2, policy((uid) -> Status.OK)) .build(); - assertThat(policy.checkAuthorizationForService(MY_UID, SERVICE1).getCode()) - .isEqualTo(Status.OK.getCode()); - assertThat(policy.checkAuthorizationForService(OTHER_UID, SERVICE1).getCode()) - .isEqualTo(Status.PERMISSION_DENIED.getCode()); - assertThat(policy.checkAuthorizationForService(MY_UID, SERVICE2).getCode()) - .isEqualTo(Status.OK.getCode()); - assertThat(policy.checkAuthorizationForService(OTHER_UID, SERVICE2).getCode()) - .isEqualTo(Status.OK.getCode()); + checkSynchronousPolicy(policy, MY_UID, SERVICE1, Code.OK); + checkSynchronousPolicy(policy, OTHER_UID, SERVICE1, Code.PERMISSION_DENIED); + checkSynchronousPolicy(policy, MY_UID, SERVICE2, Code.OK); + checkSynchronousPolicy(policy, OTHER_UID, SERVICE2, Code.OK); } @Test - public void testPerServiceAsync() { + public void testPerServiceAsync() throws Exception { policy = ServerSecurityPolicy.newBuilder() .servicePolicy(SERVICE2, asyncPolicy(uid -> { @@ -104,29 +99,30 @@ public void testPerServiceAsync() { })) .build(); - assertThat(policy.checkAuthorizationForService(MY_UID, SERVICE1).getCode()) + assertThat(policy.checkAuthorizationForServiceAsync(MY_UID, SERVICE1).get().getCode()) .isEqualTo(Status.OK.getCode()); - assertThat(policy.checkAuthorizationForService(OTHER_UID, SERVICE1).getCode()) + assertThat(policy.checkAuthorizationForServiceAsync(OTHER_UID, SERVICE1).get().getCode()) .isEqualTo(Status.PERMISSION_DENIED.getCode()); - assertThat(policy.checkAuthorizationForService(MY_UID, SERVICE2).getCode()) + assertThat(policy.checkAuthorizationForServiceAsync(MY_UID, SERVICE2).get().getCode()) .isEqualTo(Status.OK.getCode()); - assertThat(policy.checkAuthorizationForService(OTHER_UID, SERVICE2).getCode()) + assertThat(policy.checkAuthorizationForServiceAsync(OTHER_UID, SERVICE2).get().getCode()) .isEqualTo(Status.OK.getCode()); } @Test - public void testPerService_throwingExceptionAsynchronously_propagatesStatusFromException() { + public void testPerService_failedSecurityPolicyFuture_returnsAFailedFuture() { policy = ServerSecurityPolicy.newBuilder() .servicePolicy(SERVICE1, asyncPolicy(uid -> Futures - .immediateFailedFuture( - new StatusException(Status.fromCode(Status.Code.ALREADY_EXISTS))) + .immediateFailedFuture(new IllegalStateException("something went wrong")) )) .build(); - assertThat(policy.checkAuthorizationForService(MY_UID, SERVICE1).getCode()) - .isEqualTo(Status.ALREADY_EXISTS.getCode()); + ListenableFuture statusFuture = + policy.checkAuthorizationForServiceAsync(MY_UID, SERVICE1); + + assertThrows(ExecutionException.class, statusFuture::get); } @Test @@ -136,12 +132,14 @@ public void testPerServiceAsync_cancelledFuture_propagatesStatus() { .servicePolicy(SERVICE1, asyncPolicy(unused -> Futures.immediateCancelledFuture())) .build(); - assertThat(policy.checkAuthorizationForService(MY_UID, SERVICE1).getCode()) - .isEqualTo(Status.CANCELLED.getCode()); + ListenableFuture statusFuture = + policy.checkAuthorizationForServiceAsync(MY_UID, SERVICE1); + + assertThrows(CancellationException.class, statusFuture::get); } @Test - public void testPerServiceAsync_interrupted_cancelledStatus() { + public void testPerServiceAsync_interrupted_cancelledFuture() { ListeningExecutorService listeningExecutorService = MoreExecutors.listeningDecorator(Executors.newSingleThreadExecutor()); CountDownLatch unsatisfiedLatch = new CountDownLatch(1); @@ -164,15 +162,15 @@ public void testPerServiceAsync_interrupted_cancelledStatus() { return toBeInterruptedFuture; })) .build(); + ListenableFuture statusFuture = + policy.checkAuthorizationForServiceAsync(MY_UID, SERVICE1); - assertThat(policy.checkAuthorizationForService(MY_UID, SERVICE1).getCode()) - .isEqualTo(Status.CANCELLED.getCode()); - assertThat(Thread.currentThread().isInterrupted()).isTrue(); + assertThrows(InterruptedException.class, statusFuture::get); listeningExecutorService.shutdownNow(); } @Test - public void testPerServiceNoDefault() { + public void testPerServiceNoDefault() throws Exception { policy = ServerSecurityPolicy.newBuilder() .servicePolicy(SERVICE1, policy((uid) -> Status.INTERNAL)) @@ -181,26 +179,20 @@ SERVICE2, policy((uid) -> uid == OTHER_UID ? Status.OK : Status.PERMISSION_DENIE .build(); // Uses the specified policy for service1. - assertThat(policy.checkAuthorizationForService(MY_UID, SERVICE1).getCode()) - .isEqualTo(Status.INTERNAL.getCode()); - assertThat(policy.checkAuthorizationForService(OTHER_UID, SERVICE1).getCode()) - .isEqualTo(Status.INTERNAL.getCode()); + checkSynchronousPolicy(policy, MY_UID, SERVICE1, Code.INTERNAL); + checkSynchronousPolicy(policy, OTHER_UID, SERVICE1, Code.INTERNAL); // Uses the specified policy for service2. - assertThat(policy.checkAuthorizationForService(MY_UID, SERVICE2).getCode()) - .isEqualTo(Status.PERMISSION_DENIED.getCode()); - assertThat(policy.checkAuthorizationForService(OTHER_UID, SERVICE2).getCode()) - .isEqualTo(Status.OK.getCode()); + checkSynchronousPolicy(policy, MY_UID, SERVICE2, Code.PERMISSION_DENIED); + checkSynchronousPolicy(policy, OTHER_UID, SERVICE2, Code.OK); // Falls back to the default. - assertThat(policy.checkAuthorizationForService(MY_UID, SERVICE3).getCode()) - .isEqualTo(Status.OK.getCode()); - assertThat(policy.checkAuthorizationForService(OTHER_UID, SERVICE3).getCode()) - .isEqualTo(Status.PERMISSION_DENIED.getCode()); + checkSynchronousPolicy(policy, MY_UID, SERVICE3, Code.OK); + checkSynchronousPolicy(policy, OTHER_UID, SERVICE3, Code.PERMISSION_DENIED); } @Test - public void testPerServiceNoDefaultAsync() { + public void testPerServiceNoDefaultAsync() throws Exception { policy = ServerSecurityPolicy.newBuilder() .servicePolicy( @@ -224,24 +216,44 @@ SERVICE2, asyncPolicy((uid) -> { .build(); // Uses the specified policy for service1. - assertThat(policy.checkAuthorizationForService(MY_UID, SERVICE1).getCode()) + assertThat(policy.checkAuthorizationForServiceAsync(MY_UID, SERVICE1).get().getCode()) .isEqualTo(Status.INTERNAL.getCode()); - assertThat(policy.checkAuthorizationForService(OTHER_UID, SERVICE1).getCode()) + assertThat(policy.checkAuthorizationForServiceAsync(OTHER_UID, SERVICE1).get().getCode()) .isEqualTo(Status.INTERNAL.getCode()); // Uses the specified policy for service2. - assertThat(policy.checkAuthorizationForService(MY_UID, SERVICE2).getCode()) + assertThat(policy.checkAuthorizationForServiceAsync(MY_UID, SERVICE2).get().getCode()) .isEqualTo(Status.PERMISSION_DENIED.getCode()); - assertThat(policy.checkAuthorizationForService(OTHER_UID, SERVICE2).getCode()) + assertThat(policy.checkAuthorizationForServiceAsync(OTHER_UID, SERVICE2).get().getCode()) .isEqualTo(Status.OK.getCode()); // Falls back to the default. - assertThat(policy.checkAuthorizationForService(MY_UID, SERVICE3).getCode()) + assertThat(policy.checkAuthorizationForServiceAsync(MY_UID, SERVICE3).get().getCode()) .isEqualTo(Status.OK.getCode()); - assertThat(policy.checkAuthorizationForService(OTHER_UID, SERVICE3).getCode()) + assertThat(policy.checkAuthorizationForServiceAsync(OTHER_UID, SERVICE3).get().getCode()) .isEqualTo(Status.PERMISSION_DENIED.getCode()); } + /** + * Checks the resulting status of a {@link io.grpc.binder.ServerSecurityPolicy} built with a + * synchronous {@link SecurityPolicy} using both the new API and the legacy + * {@link ServerSecurityPolicy#checkAuthorizationForService}. + */ + private static void checkSynchronousPolicy( + ServerSecurityPolicy policy, + int callerUid, + String service, + Status.Code expectedCode) throws ExecutionException{ + // Legacy API; checked for backward compatibility + assertThat(policy.checkAuthorizationForService(callerUid, service).getCode()) + .isEqualTo(expectedCode); + + // New API + ListenableFuture statusFuture = + policy.checkAuthorizationForServiceAsync(callerUid, service); + assertThat(Uninterruptibles.getUninterruptibly(statusFuture).getCode()).isEqualTo(expectedCode); + } + private static SecurityPolicy policy(Function func) { return new SecurityPolicy() { @Override From 54cb5258d2ad55b5be741e99e25f48336ccfd933 Mon Sep 17 00:00:00 2001 From: Mateus Azis Date: Tue, 31 Oct 2023 10:56:31 -0700 Subject: [PATCH 02/11] - Short-circuit immediately resolved auth futures. - Drop sequential executor from PendingAuthListener. --- .../internal/BinderTransportSecurity.java | 40 ++++++- .../binder/internal/PendingAuthListener.java | 105 +++++++++++------- 2 files changed, 106 insertions(+), 39 deletions(-) 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 5eacca49192..00e6bb9bdf6 100644 --- a/binder/src/main/java/io/grpc/binder/internal/BinderTransportSecurity.java +++ b/binder/src/main/java/io/grpc/binder/internal/BinderTransportSecurity.java @@ -17,6 +17,8 @@ package io.grpc.binder.internal; import com.google.common.util.concurrent.ListenableFuture; +import com.google.common.util.concurrent.Uninterruptibles; + import io.grpc.Attributes; import io.grpc.Internal; import io.grpc.Metadata; @@ -29,7 +31,10 @@ import io.grpc.Status; import io.grpc.internal.GrpcAttributes; import io.grpc.internal.ObjectPool; + +import java.util.concurrent.CancellationException; import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.ExecutionException; import java.util.concurrent.Executor; import javax.annotation.CheckReturnValue; import javax.annotation.Nullable; @@ -87,6 +92,32 @@ private static final class ServerAuthInterceptor implements ServerInterceptor { this.executorPool = executorPool; } + /** + * @param authStatusFuture a Future that is known to be complete, for which it's safe to call + * {@link ListenableFuture#get()} without blocking the current thread. + */ + private ServerCall.Listener shortCircuitAuthFuture( + ListenableFuture authStatusFuture, + ServerCall call, + Metadata headers, + ServerCallHandler next) { + Status authStatus; + try { + authStatus = Uninterruptibles.getUninterruptibly(authStatusFuture); + } catch (ExecutionException e) { + authStatus = Status.INTERNAL.withCause(e); + } catch (CancellationException e) { + authStatus = Status.CANCELLED.withCause(e); + } + + if (authStatus.isOk()) { + return next.startCall(call, headers); + } + call.close(authStatus, new Metadata()); + return new ServerCall.Listener() { + }; + } + @Override public ServerCall.Listener interceptCall( ServerCall call, Metadata headers, ServerCallHandler next) { @@ -95,6 +126,13 @@ public ServerCall.Listener interceptCall( .get(TRANSPORT_AUTHORIZATION_STATE) .checkAuthorization(call.getMethodDescriptor()); + // Most SecurityPolicy will have synchronous implementations that provide an + // immediately-resolved Future. In that case, short-circuit to avoid unnecessary allocations + // and asynchronous code. + if (authStatusFuture.isDone()) { + return shortCircuitAuthFuture(authStatusFuture, call, headers, next); + } + return new PendingAuthListener<>(authStatusFuture, executorPool, call, headers, next); } } @@ -120,7 +158,7 @@ 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 case this cache to grow unbounded. + // created methods could cause this cache to grow unbounded. boolean useCache = method.isSampledToLocalTracing(); if (useCache) { @Nullable ListenableFuture authorization = serviceAuthorization.get(serviceName); diff --git a/binder/src/main/java/io/grpc/binder/internal/PendingAuthListener.java b/binder/src/main/java/io/grpc/binder/internal/PendingAuthListener.java index 13c58c9db01..b99d19fe81a 100644 --- a/binder/src/main/java/io/grpc/binder/internal/PendingAuthListener.java +++ b/binder/src/main/java/io/grpc/binder/internal/PendingAuthListener.java @@ -1,8 +1,8 @@ package io.grpc.binder.internal; +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 io.grpc.Metadata; import io.grpc.ServerCall; @@ -11,7 +11,11 @@ import io.grpc.Status; import io.grpc.internal.ObjectPool; +import java.util.concurrent.ConcurrentLinkedQueue; import java.util.concurrent.Executor; +import java.util.concurrent.atomic.AtomicReference; + +import javax.annotation.Nullable; /** * A {@link ServerCall.Listener} that can be returned by a {@link io.grpc.ServerInterceptor} to @@ -19,10 +23,9 @@ */ final class PendingAuthListener extends ServerCall.Listener { - private final ListenableFuture> authStatusFuture; - private final Executor sequentialExecutor; - private final ObjectPool executorPool; - private final Executor executor; + private final ConcurrentLinkedQueue> pendingSteps = + new ConcurrentLinkedQueue<>(); + private final AtomicReference> delegateRef = new AtomicReference<>(null); /** * @param authStatusFuture a ListenableFuture holding the result status of the authorization @@ -39,59 +42,85 @@ final class PendingAuthListener extends ServerCall.Listener { PendingAuthListener( ListenableFuture authStatusFuture, ObjectPool executorPool, - ServerCall call, Metadata headers, + ServerCall call, + Metadata headers, ServerCallHandler next) { - this.executorPool = executorPool; - this.executor = executorPool.getObject(); - this.authStatusFuture = Futures.transform(authStatusFuture, authStatus -> { - if (authStatus.isOk()) { - return next.startCall(call, headers); - } - call.close(authStatus, new Metadata()); - throw new IllegalStateException("Auth failed", authStatus.asException()); - }, executor); - this.sequentialExecutor = MoreExecutors.newSequentialExecutor(executor); + Executor executor = executorPool.getObject(); + + Futures.addCallback(authStatusFuture, + new FutureCallback() { + @Override + public void onSuccess(Status authStatus) { + if (authStatus.isOk()) { + delegateRef.set(next.startCall(call, headers)); + maybeRunPendingSteps(); + } else { + call.close(authStatus, new Metadata()); + } + executorPool.returnObject(executor); + } + + @Override + public void onFailure(Throwable t) { + call.close( + Status.INTERNAL.withCause(t).withDescription("Authorization future failed"), + new Metadata()); + executorPool.returnObject(executor); + } + }, executor); + } + + /** + * Runs any enqueued step in this ServerCall listener as long as the authorization check is + * complete. Otherwise, no-op and returns immediately. + */ + private void maybeRunPendingSteps() { + @Nullable Listener delegate = delegateRef.get(); + if (delegate == null) { + return; + } + + ListenerConsumer nextStep; + while ((nextStep = pendingSteps.poll()) != null) { + nextStep.accept(delegate); + } } @Override public void onCancel() { - ListenableFuture unused = Futures.transform(authStatusFuture, delegate -> { - delegate.onCancel(); - executorPool.returnObject(executor); - return null; - }, sequentialExecutor); + pendingSteps.offer(Listener::onCancel); + maybeRunPendingSteps(); } @Override public void onComplete() { - ListenableFuture unused = Futures.transform(authStatusFuture, delegate -> { - delegate.onComplete(); - executorPool.returnObject(executor); - return null; - }, sequentialExecutor); + pendingSteps.offer(Listener::onComplete); + maybeRunPendingSteps(); } @Override public void onHalfClose() { - ListenableFuture unused = Futures.transform(authStatusFuture, delegate -> { - delegate.onHalfClose(); - return null; - }, sequentialExecutor); + pendingSteps.offer(Listener::onHalfClose); + maybeRunPendingSteps(); } @Override public void onMessage(ReqT message) { - ListenableFuture unused = Futures.transform(authStatusFuture, delegate -> { - delegate.onMessage(message); - return null; - }, sequentialExecutor); + pendingSteps.offer(delegate -> delegate.onMessage(message)); + maybeRunPendingSteps(); } @Override public void onReady() { - ListenableFuture unused = Futures.transform(authStatusFuture, delegate -> { - delegate.onReady(); - return null; - }, sequentialExecutor); + pendingSteps.offer(Listener::onReady); + maybeRunPendingSteps(); + } + + /** + * Similar to Java8's {@link java.util.function.Consumer}, but redeclared in order to support + * Android SDK 21. + */ + private interface ListenerConsumer { + void accept(Listener listener); } } From 26b8cc2c56c66a28536c39b7787b9f51b504f7a3 Mon Sep 17 00:00:00 2001 From: Mateus Azis Date: Tue, 31 Oct 2023 11:02:27 -0700 Subject: [PATCH 03/11] fix indentation --- .../binder/internal/PendingAuthListener.java | 188 +++++++++--------- 1 file changed, 94 insertions(+), 94 deletions(-) diff --git a/binder/src/main/java/io/grpc/binder/internal/PendingAuthListener.java b/binder/src/main/java/io/grpc/binder/internal/PendingAuthListener.java index b99d19fe81a..5e67e0c1cb1 100644 --- a/binder/src/main/java/io/grpc/binder/internal/PendingAuthListener.java +++ b/binder/src/main/java/io/grpc/binder/internal/PendingAuthListener.java @@ -23,104 +23,104 @@ */ final class PendingAuthListener extends ServerCall.Listener { - private final ConcurrentLinkedQueue> pendingSteps = - new ConcurrentLinkedQueue<>(); - private final AtomicReference> delegateRef = new AtomicReference<>(null); - - /** - * @param authStatusFuture a ListenableFuture holding the result status of the authorization - * policy from a {@link io.grpc.binder.SecurityPolicy} or a - * {@link io.grpc.binder.AsyncSecurityPolicy}. The call only progresses - * if {@link Status#isOk()} is true. - * @param executorPool a pool that can provide at least one Executor under which the result - * of {@code authStatusFuture} can be handled, progressing the gRPC - * stages. - * @param call the 'call' parameter from {@link io.grpc.ServerInterceptor} - * @param headers the 'headers' parameter from {@link io.grpc.ServerInterceptor} - * @param next the 'next' parameter from {@link io.grpc.ServerInterceptor} - */ - PendingAuthListener( - ListenableFuture authStatusFuture, - ObjectPool executorPool, - ServerCall call, - Metadata headers, - ServerCallHandler next) { - Executor executor = executorPool.getObject(); - - Futures.addCallback(authStatusFuture, - new FutureCallback() { - @Override - public void onSuccess(Status authStatus) { - if (authStatus.isOk()) { - delegateRef.set(next.startCall(call, headers)); - maybeRunPendingSteps(); - } else { - call.close(authStatus, new Metadata()); - } - executorPool.returnObject(executor); - } - - @Override - public void onFailure(Throwable t) { - call.close( - Status.INTERNAL.withCause(t).withDescription("Authorization future failed"), - new Metadata()); - executorPool.returnObject(executor); - } - }, executor); + private final ConcurrentLinkedQueue> pendingSteps = + new ConcurrentLinkedQueue<>(); + private final AtomicReference> delegateRef = new AtomicReference<>(null); + + /** + * @param authStatusFuture a ListenableFuture holding the result status of the authorization + * policy from a {@link io.grpc.binder.SecurityPolicy} or a + * {@link io.grpc.binder.AsyncSecurityPolicy}. The call only progresses + * if {@link Status#isOk()} is true. + * @param executorPool a pool that can provide at least one Executor under which the result + * of {@code authStatusFuture} can be handled, progressing the gRPC + * stages. + * @param call the 'call' parameter from {@link io.grpc.ServerInterceptor} + * @param headers the 'headers' parameter from {@link io.grpc.ServerInterceptor} + * @param next the 'next' parameter from {@link io.grpc.ServerInterceptor} + */ + PendingAuthListener( + ListenableFuture authStatusFuture, + ObjectPool executorPool, + ServerCall call, + Metadata headers, + ServerCallHandler next) { + Executor executor = executorPool.getObject(); + Futures.addCallback( + authStatusFuture, + new FutureCallback() { + @Override + public void onSuccess(Status authStatus) { + if (authStatus.isOk()) { + delegateRef.set(next.startCall(call, headers)); + maybeRunPendingSteps(); + } else { + call.close(authStatus, new Metadata()); + } + executorPool.returnObject(executor); + } + + @Override + public void onFailure(Throwable t) { + call.close( + Status.INTERNAL.withCause(t).withDescription("Authorization future failed"), + new Metadata()); + executorPool.returnObject(executor); + } + }, executor); } - /** - * Runs any enqueued step in this ServerCall listener as long as the authorization check is - * complete. Otherwise, no-op and returns immediately. - */ - private void maybeRunPendingSteps() { - @Nullable Listener delegate = delegateRef.get(); - if (delegate == null) { - return; - } - - ListenerConsumer nextStep; - while ((nextStep = pendingSteps.poll()) != null) { - nextStep.accept(delegate); - } + /** + * Runs any enqueued step in this ServerCall listener as long as the authorization check is + * complete. Otherwise, no-op and returns immediately. + */ + private void maybeRunPendingSteps() { + @Nullable Listener delegate = delegateRef.get(); + if (delegate == null) { + return; } - @Override - public void onCancel() { - pendingSteps.offer(Listener::onCancel); - maybeRunPendingSteps(); + ListenerConsumer nextStep; + while ((nextStep = pendingSteps.poll()) != null) { + nextStep.accept(delegate); + } } - @Override - public void onComplete() { - pendingSteps.offer(Listener::onComplete); - maybeRunPendingSteps(); - } - - @Override - public void onHalfClose() { - pendingSteps.offer(Listener::onHalfClose); - maybeRunPendingSteps(); - } - - @Override - public void onMessage(ReqT message) { - pendingSteps.offer(delegate -> delegate.onMessage(message)); - maybeRunPendingSteps(); - } - - @Override - public void onReady() { - pendingSteps.offer(Listener::onReady); - maybeRunPendingSteps(); - } - - /** - * Similar to Java8's {@link java.util.function.Consumer}, but redeclared in order to support - * Android SDK 21. - */ - private interface ListenerConsumer { - void accept(Listener listener); - } + @Override + public void onCancel() { + pendingSteps.offer(Listener::onCancel); + maybeRunPendingSteps(); + } + + @Override + public void onComplete() { + pendingSteps.offer(Listener::onComplete); + maybeRunPendingSteps(); + } + + @Override + public void onHalfClose() { + pendingSteps.offer(Listener::onHalfClose); + maybeRunPendingSteps(); + } + + @Override + public void onMessage(ReqT message) { + pendingSteps.offer(delegate -> delegate.onMessage(message)); + maybeRunPendingSteps(); + } + + @Override + public void onReady() { + pendingSteps.offer(Listener::onReady); + maybeRunPendingSteps(); + } + + /** + * Similar to Java8's {@link java.util.function.Consumer}, but redeclared in order to support + * Android SDK 21. + */ + private interface ListenerConsumer { + void accept(Listener listener); + } } From 09a0190d13602d821fd7d2817a972566dcf0275d Mon Sep 17 00:00:00 2001 From: Mateus Azis Date: Tue, 7 Nov 2023 10:34:51 -0800 Subject: [PATCH 04/11] Address jdcormie's comments --- .../internal/BinderTransportSecurity.java | 19 ++++---- .../binder/internal/PendingAuthListener.java | 46 +++++++++++++------ 2 files changed, 41 insertions(+), 24 deletions(-) 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 00e6bb9bdf6..b886d0e9672 100644 --- a/binder/src/main/java/io/grpc/binder/internal/BinderTransportSecurity.java +++ b/binder/src/main/java/io/grpc/binder/internal/BinderTransportSecurity.java @@ -16,8 +16,8 @@ package io.grpc.binder.internal; +import com.google.common.util.concurrent.Futures; import com.google.common.util.concurrent.ListenableFuture; -import com.google.common.util.concurrent.Uninterruptibles; import io.grpc.Attributes; import io.grpc.Internal; @@ -32,7 +32,6 @@ import io.grpc.internal.GrpcAttributes; import io.grpc.internal.ObjectPool; -import java.util.concurrent.CancellationException; import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.ExecutionException; import java.util.concurrent.Executor; @@ -58,7 +57,9 @@ private BinderTransportSecurity() {} * @param serverBuilder The ServerBuilder being used to create the server. */ @Internal - public static void installAuthInterceptor(ServerBuilder serverBuilder, ObjectPool executorPool) { + public static void installAuthInterceptor( + ServerBuilder serverBuilder, + ObjectPool executorPool) { serverBuilder.intercept(new ServerAuthInterceptor(executorPool)); } @@ -93,21 +94,19 @@ private static final class ServerAuthInterceptor implements ServerInterceptor { } /** - * @param authStatusFuture a Future that is known to be complete, for which it's safe to call - * {@link ListenableFuture#get()} without blocking the current thread. + * @param authStatusFuture a Future that is known to be complete, i.e. + * {@link ListenableFuture#isDone()} returns true. */ - private ServerCall.Listener shortCircuitAuthFuture( + private ServerCall.Listener newServerCallListenerForDoneAuthResult( ListenableFuture authStatusFuture, ServerCall call, Metadata headers, ServerCallHandler next) { Status authStatus; try { - authStatus = Uninterruptibles.getUninterruptibly(authStatusFuture); + authStatus = Futures.getDone(authStatusFuture); } catch (ExecutionException e) { authStatus = Status.INTERNAL.withCause(e); - } catch (CancellationException e) { - authStatus = Status.CANCELLED.withCause(e); } if (authStatus.isOk()) { @@ -130,7 +129,7 @@ public ServerCall.Listener interceptCall( // immediately-resolved Future. In that case, short-circuit to avoid unnecessary allocations // and asynchronous code. if (authStatusFuture.isDone()) { - return shortCircuitAuthFuture(authStatusFuture, call, headers, next); + return newServerCallListenerForDoneAuthResult(authStatusFuture, call, headers, next); } return new PendingAuthListener<>(authStatusFuture, executorPool, call, headers, next); diff --git a/binder/src/main/java/io/grpc/binder/internal/PendingAuthListener.java b/binder/src/main/java/io/grpc/binder/internal/PendingAuthListener.java index 5e67e0c1cb1..1e8bef43f11 100644 --- a/binder/src/main/java/io/grpc/binder/internal/PendingAuthListener.java +++ b/binder/src/main/java/io/grpc/binder/internal/PendingAuthListener.java @@ -6,7 +6,6 @@ import io.grpc.Metadata; import io.grpc.ServerCall; -import io.grpc.ServerCall.Listener; import io.grpc.ServerCallHandler; import io.grpc.Status; import io.grpc.internal.ObjectPool; @@ -25,7 +24,8 @@ final class PendingAuthListener extends ServerCall.Listener { private final ConcurrentLinkedQueue> pendingSteps = new ConcurrentLinkedQueue<>(); - private final AtomicReference> delegateRef = new AtomicReference<>(null); + private final AtomicReference> delegateRef = + new AtomicReference<>(null); /** * @param authStatusFuture a ListenableFuture holding the result status of the authorization @@ -51,12 +51,26 @@ final class PendingAuthListener extends ServerCall.Listener { new FutureCallback() { @Override public void onSuccess(Status authStatus) { - if (authStatus.isOk()) { - delegateRef.set(next.startCall(call, headers)); - maybeRunPendingSteps(); - } else { + if (!authStatus.isOk()) { call.close(authStatus, new Metadata()); + executorPool.returnObject(executor); + return; } + + ServerCall.Listener delegate; + try { + delegate = next.startCall(call, headers); + } catch (Exception e) { + call.close( + Status + .INTERNAL + .withCause(e) + .withDescription("Failed to start server call after authorization check"), + new Metadata()); + return; + } + delegateRef.set(delegate); + maybeRunPendingSteps(); executorPool.returnObject(executor); } @@ -66,8 +80,8 @@ public void onFailure(Throwable t) { Status.INTERNAL.withCause(t).withDescription("Authorization future failed"), new Metadata()); executorPool.returnObject(executor); - } - }, executor); + } + }, executor); } /** @@ -75,32 +89,36 @@ public void onFailure(Throwable t) { * complete. Otherwise, no-op and returns immediately. */ private void maybeRunPendingSteps() { - @Nullable Listener delegate = delegateRef.get(); + @Nullable ServerCall.Listener delegate = delegateRef.get(); if (delegate == null) { return; } + // This section is synchronized so that no 2 threads may attempt to retrieve elements from the + // queue in order but end up executing the steps out of order. + synchronized (this) { ListenerConsumer nextStep; while ((nextStep = pendingSteps.poll()) != null) { nextStep.accept(delegate); } } + } @Override public void onCancel() { - pendingSteps.offer(Listener::onCancel); + pendingSteps.offer(ServerCall.Listener::onCancel); maybeRunPendingSteps(); } @Override public void onComplete() { - pendingSteps.offer(Listener::onComplete); + pendingSteps.offer(ServerCall.Listener::onComplete); maybeRunPendingSteps(); } @Override public void onHalfClose() { - pendingSteps.offer(Listener::onHalfClose); + pendingSteps.offer(ServerCall.Listener::onHalfClose); maybeRunPendingSteps(); } @@ -112,7 +130,7 @@ public void onMessage(ReqT message) { @Override public void onReady() { - pendingSteps.offer(Listener::onReady); + pendingSteps.offer(ServerCall.Listener::onReady); maybeRunPendingSteps(); } @@ -121,6 +139,6 @@ public void onReady() { * Android SDK 21. */ private interface ListenerConsumer { - void accept(Listener listener); + void accept(ServerCall.Listener listener); } } From 438734a131e52df10070accfef2c8c725dc9779a Mon Sep 17 00:00:00 2001 From: Mateus Azis Date: Tue, 14 Nov 2023 13:55:39 -0800 Subject: [PATCH 05/11] - Fix indentation - Factory method for PendingAuthListener. - Fix leak of executor when PendingAuthListener encountered an exception while starting the call. - Catching RuntimeException. --- .../internal/BinderTransportSecurity.java | 2 +- .../binder/internal/PendingAuthListener.java | 52 +++-- .../grpc/binder/ServerSecurityPolicyTest.java | 204 +++++++++++------- 3 files changed, 161 insertions(+), 97 deletions(-) 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 b886d0e9672..d1ecd6f4b7c 100644 --- a/binder/src/main/java/io/grpc/binder/internal/BinderTransportSecurity.java +++ b/binder/src/main/java/io/grpc/binder/internal/BinderTransportSecurity.java @@ -132,7 +132,7 @@ public ServerCall.Listener interceptCall( return newServerCallListenerForDoneAuthResult(authStatusFuture, call, headers, next); } - return new PendingAuthListener<>(authStatusFuture, executorPool, call, headers, next); + return PendingAuthListener.create(authStatusFuture, executorPool, call, headers, next); } } diff --git a/binder/src/main/java/io/grpc/binder/internal/PendingAuthListener.java b/binder/src/main/java/io/grpc/binder/internal/PendingAuthListener.java index 1e8bef43f11..262019e4246 100644 --- a/binder/src/main/java/io/grpc/binder/internal/PendingAuthListener.java +++ b/binder/src/main/java/io/grpc/binder/internal/PendingAuthListener.java @@ -39,39 +39,29 @@ final class PendingAuthListener extends ServerCall.Listener { * @param headers the 'headers' parameter from {@link io.grpc.ServerInterceptor} * @param next the 'next' parameter from {@link io.grpc.ServerInterceptor} */ - PendingAuthListener( + static PendingAuthListener create( ListenableFuture authStatusFuture, ObjectPool executorPool, ServerCall call, Metadata headers, ServerCallHandler next) { + PendingAuthListener listener = new PendingAuthListener<>(); Executor executor = executorPool.getObject(); Futures.addCallback( authStatusFuture, new FutureCallback() { @Override public void onSuccess(Status authStatus) { - if (!authStatus.isOk()) { - call.close(authStatus, new Metadata()); - executorPool.returnObject(executor); - return; - } - - ServerCall.Listener delegate; try { - delegate = next.startCall(call, headers); - } catch (Exception e) { - call.close( - Status - .INTERNAL - .withCause(e) - .withDescription("Failed to start server call after authorization check"), - new Metadata()); - return; + if (!authStatus.isOk()) { + call.close(authStatus, new Metadata()); + return; + } + + listener.startCall(call, headers, next); + } finally { + executorPool.returnObject(executor); } - delegateRef.set(delegate); - maybeRunPendingSteps(); - executorPool.returnObject(executor); } @Override @@ -82,7 +72,29 @@ public void onFailure(Throwable t) { executorPool.returnObject(executor); } }, executor); + return listener; + } + + private PendingAuthListener() {} + + private void startCall(ServerCall call, + Metadata headers, + ServerCallHandler next) { + ServerCall.Listener delegate; + try { + delegate = next.startCall(call, headers); + } catch (Exception e) { + call.close( + Status + .INTERNAL + .withCause(e) + .withDescription("Failed to start server call after authorization check"), + new Metadata()); + return; } + delegateRef.set(delegate); + maybeRunPendingSteps(); + } /** * Runs any enqueued step in this ServerCall listener as long as the authorization check is diff --git a/binder/src/test/java/io/grpc/binder/ServerSecurityPolicyTest.java b/binder/src/test/java/io/grpc/binder/ServerSecurityPolicyTest.java index 2e663075c35..8525f182b3c 100644 --- a/binder/src/test/java/io/grpc/binder/ServerSecurityPolicyTest.java +++ b/binder/src/test/java/io/grpc/binder/ServerSecurityPolicyTest.java @@ -55,22 +55,43 @@ public final class ServerSecurityPolicyTest { @Test public void testDefaultInternalOnly() throws Exception { policy = new ServerSecurityPolicy(); - checkSynchronousPolicy(policy, MY_UID, SERVICE1, Code.OK); - checkSynchronousPolicy(policy, MY_UID, SERVICE2, Code.OK); + checkAsyncPolicy(policy, MY_UID, SERVICE1, Code.OK); + checkAsyncPolicy(policy, MY_UID, SERVICE2, Code.OK); + } + + @Test + public void testDefaultInternalOnly_legacyApi() throws Exception { + policy = new ServerSecurityPolicy(); + checkLegacySynchronousPolicy(policy, MY_UID, SERVICE1, Code.OK); + checkLegacySynchronousPolicy(policy, MY_UID, SERVICE2, Code.OK); } @Test public void testInternalOnly_AnotherUid() throws Exception { policy = new ServerSecurityPolicy(); - checkSynchronousPolicy(policy, OTHER_UID, SERVICE1, Code.PERMISSION_DENIED); - checkSynchronousPolicy(policy, OTHER_UID, SERVICE2, Code.PERMISSION_DENIED); + checkAsyncPolicy(policy, OTHER_UID, SERVICE1, Code.PERMISSION_DENIED); + checkAsyncPolicy(policy, OTHER_UID, SERVICE2, Code.PERMISSION_DENIED); + } + + @Test + public void testInternalOnly_AnotherUid_legacyApi() throws Exception { + policy = new ServerSecurityPolicy(); + checkLegacySynchronousPolicy(policy, OTHER_UID, SERVICE1, Code.PERMISSION_DENIED); + checkLegacySynchronousPolicy(policy, OTHER_UID, SERVICE2, Code.PERMISSION_DENIED); } @Test public void testBuilderDefault() throws Exception { policy = ServerSecurityPolicy.newBuilder().build(); - checkSynchronousPolicy(policy, MY_UID, SERVICE1, Code.OK); - checkSynchronousPolicy(policy, OTHER_UID, SERVICE1, Code.PERMISSION_DENIED); + checkAsyncPolicy(policy, MY_UID, SERVICE1, Code.OK); + checkAsyncPolicy(policy, OTHER_UID, SERVICE1, Code.PERMISSION_DENIED); + } + + @Test + public void testBuilderDefault_legacyApi() throws Exception { + policy = ServerSecurityPolicy.newBuilder().build(); + checkLegacySynchronousPolicy(policy, MY_UID, SERVICE1, Code.OK); + checkLegacySynchronousPolicy(policy, OTHER_UID, SERVICE1, Code.PERMISSION_DENIED); } @Test @@ -80,10 +101,24 @@ public void testPerService() throws Exception { .servicePolicy(SERVICE2, policy((uid) -> Status.OK)) .build(); - checkSynchronousPolicy(policy, MY_UID, SERVICE1, Code.OK); - checkSynchronousPolicy(policy, OTHER_UID, SERVICE1, Code.PERMISSION_DENIED); - checkSynchronousPolicy(policy, MY_UID, SERVICE2, Code.OK); - checkSynchronousPolicy(policy, OTHER_UID, SERVICE2, Code.OK); + checkAsyncPolicy(policy, MY_UID, SERVICE1, Code.OK); + checkAsyncPolicy(policy, OTHER_UID, SERVICE1, Code.PERMISSION_DENIED); + checkAsyncPolicy(policy, MY_UID, SERVICE2, Code.OK); + checkAsyncPolicy(policy, OTHER_UID, SERVICE2, Code.OK); + } + + + @Test + public void testPerService_legacyApi() throws Exception { + policy = + ServerSecurityPolicy.newBuilder() + .servicePolicy(SERVICE2, policy((uid) -> Status.OK)) + .build(); + + checkLegacySynchronousPolicy(policy, MY_UID, SERVICE1, Code.OK); + checkLegacySynchronousPolicy(policy, OTHER_UID, SERVICE1, Code.PERMISSION_DENIED); + checkLegacySynchronousPolicy(policy, MY_UID, SERVICE2, Code.OK); + checkLegacySynchronousPolicy(policy, OTHER_UID, SERVICE2, Code.OK); } @Test @@ -91,22 +126,18 @@ public void testPerServiceAsync() throws Exception { policy = ServerSecurityPolicy.newBuilder() .servicePolicy(SERVICE2, asyncPolicy(uid -> { - // Add some extra future transformation to confirm that a chain - // of futures gets properly handled. - ListenableFuture dependency = Futures.immediateVoidFuture(); - return Futures - .transform(dependency, unused -> Status.OK, MoreExecutors.directExecutor()); + // Add some extra future transformation to confirm that a chain + // of futures gets properly handled. + ListenableFuture dependency = Futures.immediateVoidFuture(); + return Futures + .transform(dependency, unused -> Status.OK, MoreExecutors.directExecutor()); })) .build(); - assertThat(policy.checkAuthorizationForServiceAsync(MY_UID, SERVICE1).get().getCode()) - .isEqualTo(Status.OK.getCode()); - assertThat(policy.checkAuthorizationForServiceAsync(OTHER_UID, SERVICE1).get().getCode()) - .isEqualTo(Status.PERMISSION_DENIED.getCode()); - assertThat(policy.checkAuthorizationForServiceAsync(MY_UID, SERVICE2).get().getCode()) - .isEqualTo(Status.OK.getCode()); - assertThat(policy.checkAuthorizationForServiceAsync(OTHER_UID, SERVICE2).get().getCode()) - .isEqualTo(Status.OK.getCode()); + checkAsyncPolicy(policy, MY_UID, SERVICE1, Code.OK); + checkAsyncPolicy(policy, OTHER_UID, SERVICE1, Code.PERMISSION_DENIED); + checkAsyncPolicy(policy, MY_UID, SERVICE2, Code.OK); + checkAsyncPolicy(policy, OTHER_UID, SERVICE2, Code.OK); } @Test @@ -144,22 +175,22 @@ public void testPerServiceAsync_interrupted_cancelledFuture() { MoreExecutors.listeningDecorator(Executors.newSingleThreadExecutor()); CountDownLatch unsatisfiedLatch = new CountDownLatch(1); ListenableFuture toBeInterruptedFuture = listeningExecutorService.submit(() -> { - unsatisfiedLatch.await(); // waits forever - return null; + unsatisfiedLatch.await(); // waits forever + return null; }); CyclicBarrier barrier = new CyclicBarrier(2); Thread testThread = Thread.currentThread(); new Thread(() -> { - awaitOrFail(barrier); - testThread.interrupt(); + awaitOrFail(barrier); + testThread.interrupt(); }).start(); policy = ServerSecurityPolicy.newBuilder() .servicePolicy(SERVICE1, asyncPolicy(unused -> { - awaitOrFail(barrier); - return toBeInterruptedFuture; + awaitOrFail(barrier); + return toBeInterruptedFuture; })) .build(); ListenableFuture statusFuture = @@ -179,76 +210,97 @@ SERVICE2, policy((uid) -> uid == OTHER_UID ? Status.OK : Status.PERMISSION_DENIE .build(); // Uses the specified policy for service1. - checkSynchronousPolicy(policy, MY_UID, SERVICE1, Code.INTERNAL); - checkSynchronousPolicy(policy, OTHER_UID, SERVICE1, Code.INTERNAL); + checkAsyncPolicy(policy, MY_UID, SERVICE1, Code.INTERNAL); + checkAsyncPolicy(policy, OTHER_UID, SERVICE1, Code.INTERNAL); // Uses the specified policy for service2. - checkSynchronousPolicy(policy, MY_UID, SERVICE2, Code.PERMISSION_DENIED); - checkSynchronousPolicy(policy, OTHER_UID, SERVICE2, Code.OK); + checkAsyncPolicy(policy, MY_UID, SERVICE2, Code.PERMISSION_DENIED); + checkAsyncPolicy(policy, OTHER_UID, SERVICE2, Code.OK); // Falls back to the default. - checkSynchronousPolicy(policy, MY_UID, SERVICE3, Code.OK); - checkSynchronousPolicy(policy, OTHER_UID, SERVICE3, Code.PERMISSION_DENIED); + checkAsyncPolicy(policy, MY_UID, SERVICE3, Code.OK); + checkAsyncPolicy(policy, OTHER_UID, SERVICE3, Code.PERMISSION_DENIED); + } + @Test + public void testPerServiceNoDefault_legacyApi() throws Exception { + policy = + ServerSecurityPolicy.newBuilder() + .servicePolicy(SERVICE1, policy((uid) -> Status.INTERNAL)) + .servicePolicy( + SERVICE2, policy((uid) -> uid == OTHER_UID ? Status.OK : Status.PERMISSION_DENIED)) + .build(); + + // Uses the specified policy for service1. + checkLegacySynchronousPolicy(policy, MY_UID, SERVICE1, Code.INTERNAL); + checkLegacySynchronousPolicy(policy, OTHER_UID, SERVICE1, Code.INTERNAL); + + // Uses the specified policy for service2. + checkLegacySynchronousPolicy(policy, MY_UID, SERVICE2, Code.PERMISSION_DENIED); + checkLegacySynchronousPolicy(policy, OTHER_UID, SERVICE2, Code.OK); + + // Falls back to the default. + checkLegacySynchronousPolicy(policy, MY_UID, SERVICE3, Code.OK); + checkLegacySynchronousPolicy(policy, OTHER_UID, SERVICE3, Code.PERMISSION_DENIED); } @Test public void testPerServiceNoDefaultAsync() throws Exception { policy = - ServerSecurityPolicy.newBuilder() - .servicePolicy( - SERVICE1, - asyncPolicy((uid) -> Futures.immediateFuture(Status.INTERNAL))) - .servicePolicy( - SERVICE2, asyncPolicy((uid) -> { - // Add some extra future transformation to confirm that a chain - // of futures gets properly handled. - ListenableFuture anotherUidFuture = - Futures.immediateFuture(uid == OTHER_UID); - return Futures - .transform( - anotherUidFuture, - anotherUid -> - anotherUid - ? Status.OK - : Status.PERMISSION_DENIED, - MoreExecutors.directExecutor()); - })) - .build(); + ServerSecurityPolicy.newBuilder() + .servicePolicy( + SERVICE1, + asyncPolicy((uid) -> Futures.immediateFuture(Status.INTERNAL))) + .servicePolicy( + SERVICE2, asyncPolicy((uid) -> { + // Add some extra future transformation to confirm that a chain + // of futures gets properly handled. + ListenableFuture anotherUidFuture = + Futures.immediateFuture(uid == OTHER_UID); + return Futures + .transform( + anotherUidFuture, + anotherUid -> anotherUid ? Status.OK : Status.PERMISSION_DENIED, + MoreExecutors.directExecutor()); + })) + .build(); // Uses the specified policy for service1. - assertThat(policy.checkAuthorizationForServiceAsync(MY_UID, SERVICE1).get().getCode()) - .isEqualTo(Status.INTERNAL.getCode()); - assertThat(policy.checkAuthorizationForServiceAsync(OTHER_UID, SERVICE1).get().getCode()) - .isEqualTo(Status.INTERNAL.getCode()); + checkAsyncPolicy(policy, MY_UID, SERVICE1, Code.INTERNAL); + checkAsyncPolicy(policy, OTHER_UID, SERVICE1, Code.INTERNAL); // Uses the specified policy for service2. - assertThat(policy.checkAuthorizationForServiceAsync(MY_UID, SERVICE2).get().getCode()) - .isEqualTo(Status.PERMISSION_DENIED.getCode()); - assertThat(policy.checkAuthorizationForServiceAsync(OTHER_UID, SERVICE2).get().getCode()) - .isEqualTo(Status.OK.getCode()); + checkAsyncPolicy(policy, MY_UID, SERVICE2, Code.PERMISSION_DENIED); + checkAsyncPolicy(policy, OTHER_UID, SERVICE2, Code.OK); // Falls back to the default. - assertThat(policy.checkAuthorizationForServiceAsync(MY_UID, SERVICE3).get().getCode()) - .isEqualTo(Status.OK.getCode()); - assertThat(policy.checkAuthorizationForServiceAsync(OTHER_UID, SERVICE3).get().getCode()) - .isEqualTo(Status.PERMISSION_DENIED.getCode()); + checkAsyncPolicy(policy, MY_UID, SERVICE3, Code.OK); + checkAsyncPolicy(policy, OTHER_UID, SERVICE3, Code.PERMISSION_DENIED); } /** * Checks the resulting status of a {@link io.grpc.binder.ServerSecurityPolicy} built with a - * synchronous {@link SecurityPolicy} using both the new API and the legacy - * {@link ServerSecurityPolicy#checkAuthorizationForService}. + * using the legacy {@link ServerSecurityPolicy#checkAuthorizationForService} API for backward + * compatibility. */ - private static void checkSynchronousPolicy( + private static void checkLegacySynchronousPolicy( ServerSecurityPolicy policy, int callerUid, String service, Status.Code expectedCode) throws ExecutionException{ // Legacy API; checked for backward compatibility assertThat(policy.checkAuthorizationForService(callerUid, service).getCode()) - .isEqualTo(expectedCode); + .isEqualTo(expectedCode); + } - // New API + /** + * Checks the resulting status of a {@link io.grpc.binder.ServerSecurityPolicy} using the new, + * asynchronous API ({@link ServerSecurityPolicy#checkAuthorizationForServiceAsync}). + */ + private static void checkAsyncPolicy( + ServerSecurityPolicy policy, + int callerUid, + String service, + Status.Code expectedCode) throws ExecutionException{ ListenableFuture statusFuture = policy.checkAuthorizationForServiceAsync(callerUid, service); assertThat(Uninterruptibles.getUninterruptibly(statusFuture).getCode()).isEqualTo(expectedCode); @@ -274,12 +326,12 @@ public ListenableFuture checkAuthorizationAsync(int uid) { private static void awaitOrFail(CyclicBarrier barrier) { try { - barrier.await(); + barrier.await(); } catch (BrokenBarrierException e) { - fail(e.getMessage()); + fail(e.getMessage()); } catch (InterruptedException e) { - Thread.currentThread().interrupt(); - fail(e.getMessage()); + Thread.currentThread().interrupt(); + fail(e.getMessage()); } } } From 28eccc50276384c1debf7efa8a07ce7a9bd786c4 Mon Sep 17 00:00:00 2001 From: Mateus Azis Date: Tue, 14 Nov 2023 14:32:47 -0800 Subject: [PATCH 06/11] Exception -> RuntimeException --- .../main/java/io/grpc/binder/internal/PendingAuthListener.java | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/binder/src/main/java/io/grpc/binder/internal/PendingAuthListener.java b/binder/src/main/java/io/grpc/binder/internal/PendingAuthListener.java index 262019e4246..c41f650d794 100644 --- a/binder/src/main/java/io/grpc/binder/internal/PendingAuthListener.java +++ b/binder/src/main/java/io/grpc/binder/internal/PendingAuthListener.java @@ -83,7 +83,7 @@ private void startCall(ServerCall call, ServerCall.Listener delegate; try { delegate = next.startCall(call, headers); - } catch (Exception e) { + } catch (RuntimeException e) { call.close( Status .INTERNAL From 8cf4885f1676a915e5e8569094c0f713e6c0fbfc Mon Sep 17 00:00:00 2001 From: Mateus Azis Date: Wed, 15 Nov 2023 06:38:15 -0800 Subject: [PATCH 07/11] - Refactor test helpers. - Comment + TODO for multiple concurrent auth checks. --- .../internal/BinderTransportSecurity.java | 7 + .../grpc/binder/ServerSecurityPolicyTest.java | 143 +++++++++--------- 2 files changed, 82 insertions(+), 68 deletions(-) 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 d1ecd6f4b7c..91d47dd4acb 100644 --- a/binder/src/main/java/io/grpc/binder/internal/BinderTransportSecurity.java +++ b/binder/src/main/java/io/grpc/binder/internal/BinderTransportSecurity.java @@ -165,6 +165,13 @@ ListenableFuture checkAuthorization(MethodDescriptor method) { return authorization; } } + // 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) { diff --git a/binder/src/test/java/io/grpc/binder/ServerSecurityPolicyTest.java b/binder/src/test/java/io/grpc/binder/ServerSecurityPolicyTest.java index 8525f182b3c..291c6e88a5d 100644 --- a/binder/src/test/java/io/grpc/binder/ServerSecurityPolicyTest.java +++ b/binder/src/test/java/io/grpc/binder/ServerSecurityPolicyTest.java @@ -55,43 +55,49 @@ public final class ServerSecurityPolicyTest { @Test public void testDefaultInternalOnly() throws Exception { policy = new ServerSecurityPolicy(); - checkAsyncPolicy(policy, MY_UID, SERVICE1, Code.OK); - checkAsyncPolicy(policy, MY_UID, SERVICE2, Code.OK); + assertThat(checkAuthorizationForServiceAsync(policy, MY_UID, SERVICE1)).isEqualTo(Code.OK); + assertThat(checkAuthorizationForServiceAsync(policy, MY_UID, SERVICE2)).isEqualTo(Code.OK); } @Test - public void testDefaultInternalOnly_legacyApi() throws Exception { + public void testDefaultInternalOnly_legacyApi() { policy = new ServerSecurityPolicy(); - checkLegacySynchronousPolicy(policy, MY_UID, SERVICE1, Code.OK); - checkLegacySynchronousPolicy(policy, MY_UID, SERVICE2, Code.OK); + assertThat(policy.checkAuthorizationForService(MY_UID, SERVICE1).getCode()).isEqualTo(Code.OK); + assertThat(policy.checkAuthorizationForService(MY_UID, SERVICE2).getCode()).isEqualTo(Code.OK); } @Test public void testInternalOnly_AnotherUid() throws Exception { policy = new ServerSecurityPolicy(); - checkAsyncPolicy(policy, OTHER_UID, SERVICE1, Code.PERMISSION_DENIED); - checkAsyncPolicy(policy, OTHER_UID, SERVICE2, Code.PERMISSION_DENIED); + assertThat(checkAuthorizationForServiceAsync(policy, OTHER_UID, SERVICE1)) + .isEqualTo(Code.PERMISSION_DENIED); + assertThat(checkAuthorizationForServiceAsync(policy, OTHER_UID, SERVICE2)) + .isEqualTo(Code.PERMISSION_DENIED); } @Test - public void testInternalOnly_AnotherUid_legacyApi() throws Exception { + public void testInternalOnly_AnotherUid_legacyApi() { policy = new ServerSecurityPolicy(); - checkLegacySynchronousPolicy(policy, OTHER_UID, SERVICE1, Code.PERMISSION_DENIED); - checkLegacySynchronousPolicy(policy, OTHER_UID, SERVICE2, Code.PERMISSION_DENIED); + assertThat(policy.checkAuthorizationForService(OTHER_UID, SERVICE1).getCode()) + .isEqualTo(Code.PERMISSION_DENIED); + assertThat(policy.checkAuthorizationForService(OTHER_UID, SERVICE2).getCode()) + .isEqualTo(Code.PERMISSION_DENIED); } @Test public void testBuilderDefault() throws Exception { policy = ServerSecurityPolicy.newBuilder().build(); - checkAsyncPolicy(policy, MY_UID, SERVICE1, Code.OK); - checkAsyncPolicy(policy, OTHER_UID, SERVICE1, Code.PERMISSION_DENIED); + assertThat(checkAuthorizationForServiceAsync(policy, MY_UID, SERVICE1)).isEqualTo(Code.OK); + assertThat(checkAuthorizationForServiceAsync(policy, OTHER_UID, SERVICE1)) + .isEqualTo(Code.PERMISSION_DENIED); } @Test - public void testBuilderDefault_legacyApi() throws Exception { + public void testBuilderDefault_legacyApi() { policy = ServerSecurityPolicy.newBuilder().build(); - checkLegacySynchronousPolicy(policy, MY_UID, SERVICE1, Code.OK); - checkLegacySynchronousPolicy(policy, OTHER_UID, SERVICE1, Code.PERMISSION_DENIED); + assertThat(policy.checkAuthorizationForService(MY_UID, SERVICE1).getCode()).isEqualTo(Code.OK); + assertThat(policy.checkAuthorizationForService(OTHER_UID, SERVICE1).getCode()) + .isEqualTo(Code.PERMISSION_DENIED); } @Test @@ -101,24 +107,27 @@ public void testPerService() throws Exception { .servicePolicy(SERVICE2, policy((uid) -> Status.OK)) .build(); - checkAsyncPolicy(policy, MY_UID, SERVICE1, Code.OK); - checkAsyncPolicy(policy, OTHER_UID, SERVICE1, Code.PERMISSION_DENIED); - checkAsyncPolicy(policy, MY_UID, SERVICE2, Code.OK); - checkAsyncPolicy(policy, OTHER_UID, SERVICE2, Code.OK); + assertThat(checkAuthorizationForServiceAsync(policy, MY_UID, SERVICE1)).isEqualTo(Code.OK); + assertThat(checkAuthorizationForServiceAsync(policy, OTHER_UID, SERVICE1)) + .isEqualTo(Code.PERMISSION_DENIED); + assertThat(checkAuthorizationForServiceAsync(policy, MY_UID, SERVICE2)).isEqualTo(Code.OK); + assertThat(checkAuthorizationForServiceAsync(policy, OTHER_UID, SERVICE2)).isEqualTo(Code.OK); } @Test - public void testPerService_legacyApi() throws Exception { + public void testPerService_legacyApi() { policy = ServerSecurityPolicy.newBuilder() .servicePolicy(SERVICE2, policy((uid) -> Status.OK)) .build(); - checkLegacySynchronousPolicy(policy, MY_UID, SERVICE1, Code.OK); - checkLegacySynchronousPolicy(policy, OTHER_UID, SERVICE1, Code.PERMISSION_DENIED); - checkLegacySynchronousPolicy(policy, MY_UID, SERVICE2, Code.OK); - checkLegacySynchronousPolicy(policy, OTHER_UID, SERVICE2, Code.OK); + assertThat(policy.checkAuthorizationForService(MY_UID, SERVICE1).getCode()).isEqualTo(Code.OK); + assertThat(policy.checkAuthorizationForService(OTHER_UID, SERVICE1).getCode()) + .isEqualTo(Code.PERMISSION_DENIED); + assertThat(policy.checkAuthorizationForService(MY_UID, SERVICE2).getCode()).isEqualTo(Code.OK); + assertThat(policy.checkAuthorizationForService(OTHER_UID, SERVICE2).getCode()) + .isEqualTo(Code.OK); } @Test @@ -134,10 +143,11 @@ public void testPerServiceAsync() throws Exception { })) .build(); - checkAsyncPolicy(policy, MY_UID, SERVICE1, Code.OK); - checkAsyncPolicy(policy, OTHER_UID, SERVICE1, Code.PERMISSION_DENIED); - checkAsyncPolicy(policy, MY_UID, SERVICE2, Code.OK); - checkAsyncPolicy(policy, OTHER_UID, SERVICE2, Code.OK); + assertThat(checkAuthorizationForServiceAsync(policy, MY_UID, SERVICE1)).isEqualTo(Code.OK); + assertThat(checkAuthorizationForServiceAsync(policy, OTHER_UID, SERVICE1)) + .isEqualTo(Code.PERMISSION_DENIED); + assertThat(checkAuthorizationForServiceAsync(policy, MY_UID, SERVICE2)).isEqualTo(Code.OK); + assertThat(checkAuthorizationForServiceAsync(policy, OTHER_UID, SERVICE2)).isEqualTo(Code.OK); } @Test @@ -210,19 +220,23 @@ SERVICE2, policy((uid) -> uid == OTHER_UID ? Status.OK : Status.PERMISSION_DENIE .build(); // Uses the specified policy for service1. - checkAsyncPolicy(policy, MY_UID, SERVICE1, Code.INTERNAL); - checkAsyncPolicy(policy, OTHER_UID, SERVICE1, Code.INTERNAL); + assertThat(checkAuthorizationForServiceAsync(policy, MY_UID, SERVICE1)) + .isEqualTo(Code.INTERNAL); + assertThat(checkAuthorizationForServiceAsync(policy, OTHER_UID, SERVICE1)) + .isEqualTo(Code.INTERNAL); // Uses the specified policy for service2. - checkAsyncPolicy(policy, MY_UID, SERVICE2, Code.PERMISSION_DENIED); - checkAsyncPolicy(policy, OTHER_UID, SERVICE2, Code.OK); + assertThat(checkAuthorizationForServiceAsync(policy, MY_UID, SERVICE2)) + .isEqualTo(Code.PERMISSION_DENIED); + assertThat(checkAuthorizationForServiceAsync(policy, OTHER_UID, SERVICE2)).isEqualTo(Code.OK); // Falls back to the default. - checkAsyncPolicy(policy, MY_UID, SERVICE3, Code.OK); - checkAsyncPolicy(policy, OTHER_UID, SERVICE3, Code.PERMISSION_DENIED); + assertThat(checkAuthorizationForServiceAsync(policy, MY_UID, SERVICE3)).isEqualTo(Code.OK); + assertThat(checkAuthorizationForServiceAsync(policy, OTHER_UID, SERVICE3)) + .isEqualTo(Code.PERMISSION_DENIED); } @Test - public void testPerServiceNoDefault_legacyApi() throws Exception { + public void testPerServiceNoDefault_legacyApi() { policy = ServerSecurityPolicy.newBuilder() .servicePolicy(SERVICE1, policy((uid) -> Status.INTERNAL)) @@ -231,16 +245,22 @@ SERVICE2, policy((uid) -> uid == OTHER_UID ? Status.OK : Status.PERMISSION_DENIE .build(); // Uses the specified policy for service1. - checkLegacySynchronousPolicy(policy, MY_UID, SERVICE1, Code.INTERNAL); - checkLegacySynchronousPolicy(policy, OTHER_UID, SERVICE1, Code.INTERNAL); + assertThat(policy.checkAuthorizationForService(MY_UID, SERVICE1).getCode()) + .isEqualTo(Code.INTERNAL); + assertThat(policy.checkAuthorizationForService(OTHER_UID, SERVICE1).getCode()) + .isEqualTo(Code.INTERNAL); // Uses the specified policy for service2. - checkLegacySynchronousPolicy(policy, MY_UID, SERVICE2, Code.PERMISSION_DENIED); - checkLegacySynchronousPolicy(policy, OTHER_UID, SERVICE2, Code.OK); + assertThat(policy.checkAuthorizationForService(MY_UID, SERVICE2).getCode()) + .isEqualTo(Code.PERMISSION_DENIED); + assertThat(policy.checkAuthorizationForService(OTHER_UID, SERVICE2).getCode()) + .isEqualTo(Code.OK); // Falls back to the default. - checkLegacySynchronousPolicy(policy, MY_UID, SERVICE3, Code.OK); - checkLegacySynchronousPolicy(policy, OTHER_UID, SERVICE3, Code.PERMISSION_DENIED); + assertThat(policy.checkAuthorizationForService(MY_UID, SERVICE3).getCode()) + .isEqualTo(Code.OK); + assertThat(policy.checkAuthorizationForService(OTHER_UID, SERVICE3).getCode()) + .isEqualTo(Code.PERMISSION_DENIED); } @Test @@ -265,45 +285,32 @@ SERVICE2, asyncPolicy((uid) -> { .build(); // Uses the specified policy for service1. - checkAsyncPolicy(policy, MY_UID, SERVICE1, Code.INTERNAL); - checkAsyncPolicy(policy, OTHER_UID, SERVICE1, Code.INTERNAL); + assertThat(checkAuthorizationForServiceAsync(policy, MY_UID, SERVICE1)).isEqualTo(Code.INTERNAL); + assertThat(checkAuthorizationForServiceAsync(policy, OTHER_UID, SERVICE1)) + .isEqualTo(Code.INTERNAL); // Uses the specified policy for service2. - checkAsyncPolicy(policy, MY_UID, SERVICE2, Code.PERMISSION_DENIED); - checkAsyncPolicy(policy, OTHER_UID, SERVICE2, Code.OK); + assertThat(checkAuthorizationForServiceAsync(policy, MY_UID, SERVICE2)) + .isEqualTo(Code.PERMISSION_DENIED); + assertThat(checkAuthorizationForServiceAsync(policy, OTHER_UID, SERVICE2)).isEqualTo(Code.OK); // Falls back to the default. - checkAsyncPolicy(policy, MY_UID, SERVICE3, Code.OK); - checkAsyncPolicy(policy, OTHER_UID, SERVICE3, Code.PERMISSION_DENIED); + assertThat(checkAuthorizationForServiceAsync(policy, MY_UID, SERVICE3)).isEqualTo(Code.OK); + assertThat(checkAuthorizationForServiceAsync(policy, OTHER_UID, SERVICE3)) + .isEqualTo(Code.PERMISSION_DENIED); } /** - * Checks the resulting status of a {@link io.grpc.binder.ServerSecurityPolicy} built with a - * using the legacy {@link ServerSecurityPolicy#checkAuthorizationForService} API for backward - * compatibility. + * Shortcut for invoking {@link ServerSecurityPolicy#checkAuthorizationForServiceAsync} without + * dealing with concurrency details. Returns a {link @Status.Code} for convenience. */ - private static void checkLegacySynchronousPolicy( + private static Status.Code checkAuthorizationForServiceAsync( ServerSecurityPolicy policy, int callerUid, - String service, - Status.Code expectedCode) throws ExecutionException{ - // Legacy API; checked for backward compatibility - assertThat(policy.checkAuthorizationForService(callerUid, service).getCode()) - .isEqualTo(expectedCode); - } - - /** - * Checks the resulting status of a {@link io.grpc.binder.ServerSecurityPolicy} using the new, - * asynchronous API ({@link ServerSecurityPolicy#checkAuthorizationForServiceAsync}). - */ - private static void checkAsyncPolicy( - ServerSecurityPolicy policy, - int callerUid, - String service, - Status.Code expectedCode) throws ExecutionException{ + String service) throws ExecutionException { ListenableFuture statusFuture = policy.checkAuthorizationForServiceAsync(callerUid, service); - assertThat(Uninterruptibles.getUninterruptibly(statusFuture).getCode()).isEqualTo(expectedCode); + return Uninterruptibles.getUninterruptibly(statusFuture).getCode(); } private static SecurityPolicy policy(Function func) { From 889c524737a60b87e3d918e4ffd272b212c2d45f Mon Sep 17 00:00:00 2001 From: Mateus Azis Date: Thu, 16 Nov 2023 10:45:56 -0800 Subject: [PATCH 08/11] Move ListenableFuture handling logic to BinderTransportSecurity --- .../internal/BinderTransportSecurity.java | 30 +++++++++- .../binder/internal/PendingAuthListener.java | 58 +------------------ 2 files changed, 31 insertions(+), 57 deletions(-) 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 91d47dd4acb..602d7423175 100644 --- a/binder/src/main/java/io/grpc/binder/internal/BinderTransportSecurity.java +++ b/binder/src/main/java/io/grpc/binder/internal/BinderTransportSecurity.java @@ -16,6 +16,7 @@ package io.grpc.binder.internal; +import com.google.common.util.concurrent.FutureCallback; import com.google.common.util.concurrent.Futures; import com.google.common.util.concurrent.ListenableFuture; @@ -132,7 +133,34 @@ public ServerCall.Listener interceptCall( return newServerCallListenerForDoneAuthResult(authStatusFuture, call, headers, next); } - return PendingAuthListener.create(authStatusFuture, executorPool, call, headers, next); + PendingAuthListener listener = new PendingAuthListener<>(); + Executor executor = executorPool.getObject(); + Futures.addCallback( + authStatusFuture, + new FutureCallback() { + @Override + public void onSuccess(Status authStatus) { + try { + if (!authStatus.isOk()) { + call.close(authStatus, new Metadata()); + return; + } + + listener.startCall(call, headers, next); + } finally { + executorPool.returnObject(executor); + } + } + + @Override + public void onFailure(Throwable t) { + call.close( + Status.INTERNAL.withCause(t).withDescription("Authorization future failed"), + new Metadata()); + executorPool.returnObject(executor); + } + }, executor); + return listener; } } diff --git a/binder/src/main/java/io/grpc/binder/internal/PendingAuthListener.java b/binder/src/main/java/io/grpc/binder/internal/PendingAuthListener.java index c41f650d794..cdafc9c9191 100644 --- a/binder/src/main/java/io/grpc/binder/internal/PendingAuthListener.java +++ b/binder/src/main/java/io/grpc/binder/internal/PendingAuthListener.java @@ -1,17 +1,11 @@ package io.grpc.binder.internal; -import com.google.common.util.concurrent.FutureCallback; -import com.google.common.util.concurrent.Futures; -import com.google.common.util.concurrent.ListenableFuture; - import io.grpc.Metadata; import io.grpc.ServerCall; import io.grpc.ServerCallHandler; import io.grpc.Status; -import io.grpc.internal.ObjectPool; import java.util.concurrent.ConcurrentLinkedQueue; -import java.util.concurrent.Executor; import java.util.concurrent.atomic.AtomicReference; import javax.annotation.Nullable; @@ -27,57 +21,9 @@ final class PendingAuthListener extends ServerCall.Listener { private final AtomicReference> delegateRef = new AtomicReference<>(null); - /** - * @param authStatusFuture a ListenableFuture holding the result status of the authorization - * policy from a {@link io.grpc.binder.SecurityPolicy} or a - * {@link io.grpc.binder.AsyncSecurityPolicy}. The call only progresses - * if {@link Status#isOk()} is true. - * @param executorPool a pool that can provide at least one Executor under which the result - * of {@code authStatusFuture} can be handled, progressing the gRPC - * stages. - * @param call the 'call' parameter from {@link io.grpc.ServerInterceptor} - * @param headers the 'headers' parameter from {@link io.grpc.ServerInterceptor} - * @param next the 'next' parameter from {@link io.grpc.ServerInterceptor} - */ - static PendingAuthListener create( - ListenableFuture authStatusFuture, - ObjectPool executorPool, - ServerCall call, - Metadata headers, - ServerCallHandler next) { - PendingAuthListener listener = new PendingAuthListener<>(); - Executor executor = executorPool.getObject(); - Futures.addCallback( - authStatusFuture, - new FutureCallback() { - @Override - public void onSuccess(Status authStatus) { - try { - if (!authStatus.isOk()) { - call.close(authStatus, new Metadata()); - return; - } - - listener.startCall(call, headers, next); - } finally { - executorPool.returnObject(executor); - } - } - - @Override - public void onFailure(Throwable t) { - call.close( - Status.INTERNAL.withCause(t).withDescription("Authorization future failed"), - new Metadata()); - executorPool.returnObject(executor); - } - }, executor); - return listener; - } - - private PendingAuthListener() {} + PendingAuthListener() {} - private void startCall(ServerCall call, + void startCall(ServerCall call, Metadata headers, ServerCallHandler next) { ServerCall.Listener delegate; From 6a7fd67b62be3e1552e15eceebfdeed52f73193e Mon Sep 17 00:00:00 2001 From: Mateus Azis Date: Tue, 28 Nov 2023 13:41:47 -0800 Subject: [PATCH 09/11] Move the lifecycle management of the executor to the BinderServer. --- .../binder/internal/BinderTransportTest.java | 3 +- .../io/grpc/binder/BinderServerBuilder.java | 14 +++++++-- .../io/grpc/binder/internal/BinderServer.java | 19 +++++++++++- .../internal/BinderTransportSecurity.java | 30 +++++++------------ 4 files changed, 43 insertions(+), 23 deletions(-) diff --git a/binder/src/androidTest/java/io/grpc/binder/internal/BinderTransportTest.java b/binder/src/androidTest/java/io/grpc/binder/internal/BinderTransportTest.java index c06bf88955e..3bce7b36a03 100644 --- a/binder/src/androidTest/java/io/grpc/binder/internal/BinderTransportTest.java +++ b/binder/src/androidTest/java/io/grpc/binder/internal/BinderTransportTest.java @@ -71,7 +71,8 @@ protected InternalServer newServer(List streamTracer executorServicePool, streamTracerFactories, BinderInternal.createPolicyChecker(SecurityPolicies.serverInternalOnly()), - InboundParcelablePolicy.DEFAULT); + InboundParcelablePolicy.DEFAULT, + /* closeableExecutor=*/ null); HostServices.configureService(addr, HostServices.serviceParamsBuilder() diff --git a/binder/src/main/java/io/grpc/binder/BinderServerBuilder.java b/binder/src/main/java/io/grpc/binder/BinderServerBuilder.java index fe3d71ebf1d..776f152acf5 100644 --- a/binder/src/main/java/io/grpc/binder/BinderServerBuilder.java +++ b/binder/src/main/java/io/grpc/binder/BinderServerBuilder.java @@ -33,9 +33,14 @@ import io.grpc.internal.ServerImplBuilder; import io.grpc.internal.ObjectPool; import io.grpc.internal.SharedResourcePool; + +import java.io.Closeable; import java.io.File; +import java.util.concurrent.Executor; import java.util.concurrent.ScheduledExecutorService; +import javax.annotation.Nullable; + /** * Builder for a server that services requests from an Android Service. */ @@ -72,6 +77,7 @@ public static BinderServerBuilder forPort(int port) { private ServerSecurityPolicy securityPolicy; private InboundParcelablePolicy inboundParcelablePolicy; private boolean isBuilt; + @Nullable private Closeable closeableExecutor = null; private BinderServerBuilder( AndroidComponentAddress listenAddress, @@ -85,7 +91,8 @@ private BinderServerBuilder( schedulerPool, streamTracerFactories, BinderInternal.createPolicyChecker(securityPolicy), - inboundParcelablePolicy); + inboundParcelablePolicy, + closeableExecutor); BinderInternal.setIBinder(binderReceiver, server.getHostBinder()); return server; }); @@ -171,7 +178,10 @@ public Server build() { checkState(!isBuilt, "BinderServerBuilder can only be used to build one server instance."); isBuilt = true; // We install the security interceptor last, so it's closest to the transport. - BinderTransportSecurity.installAuthInterceptor(this, serverImplBuilder.getExecutorPool()); + ObjectPool executorPool = serverImplBuilder.getExecutorPool(); + Executor executor = executorPool.getObject(); + BinderTransportSecurity.installAuthInterceptor(this, executor); + closeableExecutor = () -> executorPool.returnObject(executor); return super.build(); } } diff --git a/binder/src/main/java/io/grpc/binder/internal/BinderServer.java b/binder/src/main/java/io/grpc/binder/internal/BinderServer.java index e72f2851b29..4c03803861f 100644 --- a/binder/src/main/java/io/grpc/binder/internal/BinderServer.java +++ b/binder/src/main/java/io/grpc/binder/internal/BinderServer.java @@ -35,6 +35,8 @@ import io.grpc.internal.ObjectPool; import io.grpc.internal.ServerListener; import io.grpc.internal.SharedResourceHolder; + +import java.io.Closeable; import java.io.IOException; import java.net.SocketAddress; import java.util.List; @@ -60,6 +62,7 @@ public final class BinderServer implements InternalServer, LeakSafeOneWayBinder. private final LeakSafeOneWayBinder hostServiceBinder; private final BinderTransportSecurity.ServerPolicyChecker serverPolicyChecker; private final InboundParcelablePolicy inboundParcelablePolicy; + @Nullable private final Closeable closeableExecutor; @GuardedBy("this") private ServerListener listener; @@ -70,18 +73,24 @@ public final class BinderServer implements InternalServer, LeakSafeOneWayBinder. @GuardedBy("this") private boolean shutdown; + /** + * @param closeableExecutor if non-null, represents an Executor from an {@link ObjectPool} that + * should be returned once the server shuts down. + */ public BinderServer( AndroidComponentAddress listenAddress, ObjectPool executorServicePool, List streamTracerFactories, BinderTransportSecurity.ServerPolicyChecker serverPolicyChecker, - InboundParcelablePolicy inboundParcelablePolicy) { + InboundParcelablePolicy inboundParcelablePolicy, + @Nullable Closeable closeableExecutor) { this.listenAddress = listenAddress; this.executorServicePool = executorServicePool; this.streamTracerFactories = ImmutableList.copyOf(checkNotNull(streamTracerFactories, "streamTracerFactories")); this.serverPolicyChecker = checkNotNull(serverPolicyChecker, "serverPolicyChecker"); this.inboundParcelablePolicy = inboundParcelablePolicy; + this.closeableExecutor = closeableExecutor; hostServiceBinder = new LeakSafeOneWayBinder(this); } @@ -125,6 +134,14 @@ public synchronized void shutdown() { hostServiceBinder.detach(); listener.serverShutdown(); executorService = executorServicePool.returnObject(executorService); + try { + if (closeableExecutor != null) { + closeableExecutor.close(); + } + } catch (IOException e) { + // Impossible. This should be just a call to "ObjectPool.return()". + throw new AssertionError(e); + } } } 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 602d7423175..5e1a77e21bf 100644 --- a/binder/src/main/java/io/grpc/binder/internal/BinderTransportSecurity.java +++ b/binder/src/main/java/io/grpc/binder/internal/BinderTransportSecurity.java @@ -31,7 +31,6 @@ import io.grpc.ServerInterceptor; import io.grpc.Status; import io.grpc.internal.GrpcAttributes; -import io.grpc.internal.ObjectPool; import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.ExecutionException; @@ -56,12 +55,11 @@ private BinderTransportSecurity() {} * Install a security policy on an about-to-be created server. * * @param serverBuilder The ServerBuilder being used to create the server. + * @param executor The executor in which the authorization result will be handled. */ @Internal - public static void installAuthInterceptor( - ServerBuilder serverBuilder, - ObjectPool executorPool) { - serverBuilder.intercept(new ServerAuthInterceptor(executorPool)); + public static void installAuthInterceptor(ServerBuilder serverBuilder, Executor executor) { + serverBuilder.intercept(new ServerAuthInterceptor(executor)); } /** @@ -88,10 +86,10 @@ public static void attachAuthAttrs( */ private static final class ServerAuthInterceptor implements ServerInterceptor { - private final ObjectPool executorPool; + private final Executor executor; - ServerAuthInterceptor(ObjectPool executorPool) { - this.executorPool = executorPool; + ServerAuthInterceptor(Executor executor) { + this.executor = executor; } /** @@ -134,22 +132,17 @@ public ServerCall.Listener interceptCall( } PendingAuthListener listener = new PendingAuthListener<>(); - Executor executor = executorPool.getObject(); Futures.addCallback( authStatusFuture, new FutureCallback() { @Override public void onSuccess(Status authStatus) { - try { - if (!authStatus.isOk()) { - call.close(authStatus, new Metadata()); - return; - } - - listener.startCall(call, headers, next); - } finally { - executorPool.returnObject(executor); + if (!authStatus.isOk()) { + call.close(authStatus, new Metadata()); + return; } + + listener.startCall(call, headers, next); } @Override @@ -157,7 +150,6 @@ public void onFailure(Throwable t) { call.close( Status.INTERNAL.withCause(t).withDescription("Authorization future failed"), new Metadata()); - executorPool.returnObject(executor); } }, executor); return listener; From 555932068a4d2e3244f38be18c2b4e094b569a49 Mon Sep 17 00:00:00 2001 From: Mateus Azis Date: Wed, 29 Nov 2023 15:33:46 -0800 Subject: [PATCH 10/11] - revert formatting changes in ServerSecurityPolicyTest - define BinderTransportSecurity.ShutdownListener interface - take a non-close executor closeable in BinderServer's constructor --- .../binder/internal/BinderTransportTest.java | 2 +- .../io/grpc/binder/BinderServerBuilder.java | 7 +- .../io/grpc/binder/internal/BinderServer.java | 19 +- .../internal/BinderTransportSecurity.java | 86 +++++---- .../grpc/binder/ServerSecurityPolicyTest.java | 171 ++++++++++-------- 5 files changed, 154 insertions(+), 131 deletions(-) diff --git a/binder/src/androidTest/java/io/grpc/binder/internal/BinderTransportTest.java b/binder/src/androidTest/java/io/grpc/binder/internal/BinderTransportTest.java index 3bce7b36a03..53a724fc7e0 100644 --- a/binder/src/androidTest/java/io/grpc/binder/internal/BinderTransportTest.java +++ b/binder/src/androidTest/java/io/grpc/binder/internal/BinderTransportTest.java @@ -72,7 +72,7 @@ protected InternalServer newServer(List streamTracer streamTracerFactories, BinderInternal.createPolicyChecker(SecurityPolicies.serverInternalOnly()), InboundParcelablePolicy.DEFAULT, - /* closeableExecutor=*/ null); + /* shutdownListener=*/ () -> {}); HostServices.configureService(addr, HostServices.serviceParamsBuilder() diff --git a/binder/src/main/java/io/grpc/binder/BinderServerBuilder.java b/binder/src/main/java/io/grpc/binder/BinderServerBuilder.java index 776f152acf5..158f7947ee8 100644 --- a/binder/src/main/java/io/grpc/binder/BinderServerBuilder.java +++ b/binder/src/main/java/io/grpc/binder/BinderServerBuilder.java @@ -77,7 +77,7 @@ public static BinderServerBuilder forPort(int port) { private ServerSecurityPolicy securityPolicy; private InboundParcelablePolicy inboundParcelablePolicy; private boolean isBuilt; - @Nullable private Closeable closeableExecutor = null; + @Nullable private BinderTransportSecurity.ShutdownListener shutdownListener = null; private BinderServerBuilder( AndroidComponentAddress listenAddress, @@ -92,7 +92,8 @@ private BinderServerBuilder( streamTracerFactories, BinderInternal.createPolicyChecker(securityPolicy), inboundParcelablePolicy, - closeableExecutor); + // 'shutdownListener' should have been set by build() + checkNotNull(shutdownListener)); BinderInternal.setIBinder(binderReceiver, server.getHostBinder()); return server; }); @@ -181,7 +182,7 @@ public Server build() { ObjectPool executorPool = serverImplBuilder.getExecutorPool(); Executor executor = executorPool.getObject(); BinderTransportSecurity.installAuthInterceptor(this, executor); - closeableExecutor = () -> executorPool.returnObject(executor); + shutdownListener = () -> executorPool.returnObject(executor); return super.build(); } } diff --git a/binder/src/main/java/io/grpc/binder/internal/BinderServer.java b/binder/src/main/java/io/grpc/binder/internal/BinderServer.java index 4c03803861f..43a6dfcae47 100644 --- a/binder/src/main/java/io/grpc/binder/internal/BinderServer.java +++ b/binder/src/main/java/io/grpc/binder/internal/BinderServer.java @@ -62,7 +62,7 @@ public final class BinderServer implements InternalServer, LeakSafeOneWayBinder. private final LeakSafeOneWayBinder hostServiceBinder; private final BinderTransportSecurity.ServerPolicyChecker serverPolicyChecker; private final InboundParcelablePolicy inboundParcelablePolicy; - @Nullable private final Closeable closeableExecutor; + private final BinderTransportSecurity.ShutdownListener shutdownListener; @GuardedBy("this") private ServerListener listener; @@ -74,8 +74,8 @@ public final class BinderServer implements InternalServer, LeakSafeOneWayBinder. private boolean shutdown; /** - * @param closeableExecutor if non-null, represents an Executor from an {@link ObjectPool} that - * should be returned once the server shuts down. + * @param shutdownListener represents resources that should be cleaned up once the server shuts + * down. */ public BinderServer( AndroidComponentAddress listenAddress, @@ -83,14 +83,14 @@ public BinderServer( List streamTracerFactories, BinderTransportSecurity.ServerPolicyChecker serverPolicyChecker, InboundParcelablePolicy inboundParcelablePolicy, - @Nullable Closeable closeableExecutor) { + BinderTransportSecurity.ShutdownListener shutdownListener) { this.listenAddress = listenAddress; this.executorServicePool = executorServicePool; this.streamTracerFactories = ImmutableList.copyOf(checkNotNull(streamTracerFactories, "streamTracerFactories")); this.serverPolicyChecker = checkNotNull(serverPolicyChecker, "serverPolicyChecker"); this.inboundParcelablePolicy = inboundParcelablePolicy; - this.closeableExecutor = closeableExecutor; + this.shutdownListener = shutdownListener; hostServiceBinder = new LeakSafeOneWayBinder(this); } @@ -134,14 +134,7 @@ public synchronized void shutdown() { hostServiceBinder.detach(); listener.serverShutdown(); executorService = executorServicePool.returnObject(executorService); - try { - if (closeableExecutor != null) { - closeableExecutor.close(); - } - } catch (IOException e) { - // Impossible. This should be just a call to "ObjectPool.return()". - throw new AssertionError(e); - } + shutdownListener.onShutdown(); } } 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 5e1a77e21bf..c5e36ca9765 100644 --- a/binder/src/main/java/io/grpc/binder/internal/BinderTransportSecurity.java +++ b/binder/src/main/java/io/grpc/binder/internal/BinderTransportSecurity.java @@ -96,24 +96,33 @@ private static final class ServerAuthInterceptor implements ServerInterceptor { * @param authStatusFuture a Future that is known to be complete, i.e. * {@link ListenableFuture#isDone()} returns true. */ - private ServerCall.Listener newServerCallListenerForDoneAuthResult( + private ServerCall.Listener newServerCallListenerForPendingAuthResult( ListenableFuture authStatusFuture, ServerCall call, Metadata headers, ServerCallHandler next) { - Status authStatus; - try { - authStatus = Futures.getDone(authStatusFuture); - } catch (ExecutionException e) { - authStatus = Status.INTERNAL.withCause(e); - } - - if (authStatus.isOk()) { - return next.startCall(call, headers); - } - call.close(authStatus, new Metadata()); - return new ServerCall.Listener() { - }; + PendingAuthListener listener = new PendingAuthListener<>(); + Futures.addCallback( + authStatusFuture, + new FutureCallback() { + @Override + public void onSuccess(Status authStatus) { + if (!authStatus.isOk()) { + call.close(authStatus, new Metadata()); + return; + } + + listener.startCall(call, headers, next); + } + + @Override + public void onFailure(Throwable t) { + call.close( + Status.INTERNAL.withCause(t).withDescription("Authorization future failed"), + new Metadata()); + } + }, executor); + return listener; } @Override @@ -126,33 +135,24 @@ public ServerCall.Listener interceptCall( // Most SecurityPolicy will have synchronous implementations that provide an // immediately-resolved Future. In that case, short-circuit to avoid unnecessary allocations - // and asynchronous code. - if (authStatusFuture.isDone()) { - return newServerCallListenerForDoneAuthResult(authStatusFuture, call, headers, next); + // and asynchronous code if the authorization result is already present. + if (!authStatusFuture.isDone()) { + return newServerCallListenerForPendingAuthResult(authStatusFuture, call, headers, next); } - PendingAuthListener listener = new PendingAuthListener<>(); - Futures.addCallback( - authStatusFuture, - new FutureCallback() { - @Override - public void onSuccess(Status authStatus) { - if (!authStatus.isOk()) { - call.close(authStatus, new Metadata()); - return; - } - - listener.startCall(call, headers, next); - } - - @Override - public void onFailure(Throwable t) { - call.close( - Status.INTERNAL.withCause(t).withDescription("Authorization future failed"), - new Metadata()); - } - }, executor); - return listener; + Status authStatus; + try { + authStatus = Futures.getDone(authStatusFuture); + } catch (ExecutionException e) { + authStatus = Status.INTERNAL.withCause(e); + } + + if (authStatus.isOk()) { + return next.startCall(call, headers); + } else { + call.close(authStatus, new Metadata()); + return new ServerCall.Listener() {}; + } } } @@ -224,4 +224,12 @@ public interface ServerPolicyChecker { */ ListenableFuture checkAuthorizationForServiceAsync(int uid, String serviceName); } + + /** + * A listener invoked when the {@link io.grpc.binder.internal.BinderServer} shuts down, allowing + * resources to be potentially cleaned up. + */ + public interface ShutdownListener { + void onShutdown(); + } } diff --git a/binder/src/test/java/io/grpc/binder/ServerSecurityPolicyTest.java b/binder/src/test/java/io/grpc/binder/ServerSecurityPolicyTest.java index 291c6e88a5d..fb7e8e05566 100644 --- a/binder/src/test/java/io/grpc/binder/ServerSecurityPolicyTest.java +++ b/binder/src/test/java/io/grpc/binder/ServerSecurityPolicyTest.java @@ -55,49 +55,55 @@ public final class ServerSecurityPolicyTest { @Test public void testDefaultInternalOnly() throws Exception { policy = new ServerSecurityPolicy(); - assertThat(checkAuthorizationForServiceAsync(policy, MY_UID, SERVICE1)).isEqualTo(Code.OK); - assertThat(checkAuthorizationForServiceAsync(policy, MY_UID, SERVICE2)).isEqualTo(Code.OK); + assertThat(checkAuthorizationForServiceAsync(policy, MY_UID, SERVICE1)) + .isEqualTo(Status.OK.getCode()); + assertThat(checkAuthorizationForServiceAsync(policy, MY_UID, SERVICE2)) + .isEqualTo(Status.OK.getCode()); } @Test public void testDefaultInternalOnly_legacyApi() { policy = new ServerSecurityPolicy(); - assertThat(policy.checkAuthorizationForService(MY_UID, SERVICE1).getCode()).isEqualTo(Code.OK); - assertThat(policy.checkAuthorizationForService(MY_UID, SERVICE2).getCode()).isEqualTo(Code.OK); + assertThat(policy.checkAuthorizationForService(MY_UID, SERVICE1).getCode()) + .isEqualTo(Status.OK.getCode()); + assertThat(policy.checkAuthorizationForService(MY_UID, SERVICE2).getCode()) + .isEqualTo(Status.OK.getCode()); } @Test public void testInternalOnly_AnotherUid() throws Exception { policy = new ServerSecurityPolicy(); assertThat(checkAuthorizationForServiceAsync(policy, OTHER_UID, SERVICE1)) - .isEqualTo(Code.PERMISSION_DENIED); + .isEqualTo(Status.PERMISSION_DENIED.getCode()); assertThat(checkAuthorizationForServiceAsync(policy, OTHER_UID, SERVICE2)) - .isEqualTo(Code.PERMISSION_DENIED); + .isEqualTo(Status.PERMISSION_DENIED.getCode()); } @Test public void testInternalOnly_AnotherUid_legacyApi() { policy = new ServerSecurityPolicy(); assertThat(policy.checkAuthorizationForService(OTHER_UID, SERVICE1).getCode()) - .isEqualTo(Code.PERMISSION_DENIED); + .isEqualTo(Status.PERMISSION_DENIED.getCode()); assertThat(policy.checkAuthorizationForService(OTHER_UID, SERVICE2).getCode()) - .isEqualTo(Code.PERMISSION_DENIED); + .isEqualTo(Status.PERMISSION_DENIED.getCode()); } @Test public void testBuilderDefault() throws Exception { policy = ServerSecurityPolicy.newBuilder().build(); - assertThat(checkAuthorizationForServiceAsync(policy, MY_UID, SERVICE1)).isEqualTo(Code.OK); + assertThat(checkAuthorizationForServiceAsync(policy, MY_UID, SERVICE1)) + .isEqualTo(Status.OK.getCode()); assertThat(checkAuthorizationForServiceAsync(policy, OTHER_UID, SERVICE1)) - .isEqualTo(Code.PERMISSION_DENIED); + .isEqualTo(Status.PERMISSION_DENIED.getCode()); } @Test public void testBuilderDefault_legacyApi() { policy = ServerSecurityPolicy.newBuilder().build(); - assertThat(policy.checkAuthorizationForService(MY_UID, SERVICE1).getCode()).isEqualTo(Code.OK); + assertThat(policy.checkAuthorizationForService(MY_UID, SERVICE1).getCode()) + .isEqualTo(Status.OK.getCode()); assertThat(policy.checkAuthorizationForService(OTHER_UID, SERVICE1).getCode()) - .isEqualTo(Code.PERMISSION_DENIED); + .isEqualTo(Status.PERMISSION_DENIED.getCode()); } @Test @@ -107,11 +113,14 @@ public void testPerService() throws Exception { .servicePolicy(SERVICE2, policy((uid) -> Status.OK)) .build(); - assertThat(checkAuthorizationForServiceAsync(policy, MY_UID, SERVICE1)).isEqualTo(Code.OK); + assertThat(checkAuthorizationForServiceAsync(policy, MY_UID, SERVICE1)) + .isEqualTo(Status.OK.getCode()); assertThat(checkAuthorizationForServiceAsync(policy, OTHER_UID, SERVICE1)) - .isEqualTo(Code.PERMISSION_DENIED); - assertThat(checkAuthorizationForServiceAsync(policy, MY_UID, SERVICE2)).isEqualTo(Code.OK); - assertThat(checkAuthorizationForServiceAsync(policy, OTHER_UID, SERVICE2)).isEqualTo(Code.OK); + .isEqualTo(Status.PERMISSION_DENIED.getCode()); + assertThat(checkAuthorizationForServiceAsync(policy, MY_UID, SERVICE2)) + .isEqualTo(Status.OK.getCode()); + assertThat(checkAuthorizationForServiceAsync(policy, OTHER_UID, SERVICE2)) + .isEqualTo(Status.OK.getCode()); } @@ -122,12 +131,14 @@ public void testPerService_legacyApi() { .servicePolicy(SERVICE2, policy((uid) -> Status.OK)) .build(); - assertThat(policy.checkAuthorizationForService(MY_UID, SERVICE1).getCode()).isEqualTo(Code.OK); + assertThat(policy.checkAuthorizationForService(MY_UID, SERVICE1).getCode()) + .isEqualTo(Status.OK.getCode()); assertThat(policy.checkAuthorizationForService(OTHER_UID, SERVICE1).getCode()) - .isEqualTo(Code.PERMISSION_DENIED); - assertThat(policy.checkAuthorizationForService(MY_UID, SERVICE2).getCode()).isEqualTo(Code.OK); + .isEqualTo(Status.PERMISSION_DENIED.getCode()); + assertThat(policy.checkAuthorizationForService(MY_UID, SERVICE2).getCode()) + .isEqualTo(Status.OK.getCode()); assertThat(policy.checkAuthorizationForService(OTHER_UID, SERVICE2).getCode()) - .isEqualTo(Code.OK); + .isEqualTo(Status.OK.getCode()); } @Test @@ -135,19 +146,22 @@ public void testPerServiceAsync() throws Exception { policy = ServerSecurityPolicy.newBuilder() .servicePolicy(SERVICE2, asyncPolicy(uid -> { - // Add some extra future transformation to confirm that a chain - // of futures gets properly handled. - ListenableFuture dependency = Futures.immediateVoidFuture(); - return Futures - .transform(dependency, unused -> Status.OK, MoreExecutors.directExecutor()); + // Add some extra future transformation to confirm that a chain + // of futures gets properly handled. + ListenableFuture dependency = Futures.immediateVoidFuture(); + return Futures + .transform(dependency, unused -> Status.OK, MoreExecutors.directExecutor()); })) .build(); - assertThat(checkAuthorizationForServiceAsync(policy, MY_UID, SERVICE1)).isEqualTo(Code.OK); + assertThat(checkAuthorizationForServiceAsync(policy, MY_UID, SERVICE1)) + .isEqualTo(Status.OK.getCode()); assertThat(checkAuthorizationForServiceAsync(policy, OTHER_UID, SERVICE1)) - .isEqualTo(Code.PERMISSION_DENIED); - assertThat(checkAuthorizationForServiceAsync(policy, MY_UID, SERVICE2)).isEqualTo(Code.OK); - assertThat(checkAuthorizationForServiceAsync(policy, OTHER_UID, SERVICE2)).isEqualTo(Code.OK); + .isEqualTo(Status.PERMISSION_DENIED.getCode()); + assertThat(checkAuthorizationForServiceAsync(policy, MY_UID, SERVICE2)) + .isEqualTo(Status.OK.getCode()); + assertThat(checkAuthorizationForServiceAsync(policy, OTHER_UID, SERVICE2)) + .isEqualTo(Status.OK.getCode()); } @Test @@ -156,7 +170,8 @@ public void testPerService_failedSecurityPolicyFuture_returnsAFailedFuture() { ServerSecurityPolicy.newBuilder() .servicePolicy(SERVICE1, asyncPolicy(uid -> Futures - .immediateFailedFuture(new IllegalStateException("something went wrong")) + .immediateFailedFuture( + new IllegalStateException("something went wrong")) )) .build(); @@ -185,22 +200,22 @@ public void testPerServiceAsync_interrupted_cancelledFuture() { MoreExecutors.listeningDecorator(Executors.newSingleThreadExecutor()); CountDownLatch unsatisfiedLatch = new CountDownLatch(1); ListenableFuture toBeInterruptedFuture = listeningExecutorService.submit(() -> { - unsatisfiedLatch.await(); // waits forever - return null; + unsatisfiedLatch.await(); // waits forever + return null; }); CyclicBarrier barrier = new CyclicBarrier(2); Thread testThread = Thread.currentThread(); new Thread(() -> { - awaitOrFail(barrier); - testThread.interrupt(); + awaitOrFail(barrier); + testThread.interrupt(); }).start(); policy = ServerSecurityPolicy.newBuilder() .servicePolicy(SERVICE1, asyncPolicy(unused -> { - awaitOrFail(barrier); - return toBeInterruptedFuture; + awaitOrFail(barrier); + return toBeInterruptedFuture; })) .build(); ListenableFuture statusFuture = @@ -221,19 +236,19 @@ SERVICE2, policy((uid) -> uid == OTHER_UID ? Status.OK : Status.PERMISSION_DENIE // Uses the specified policy for service1. assertThat(checkAuthorizationForServiceAsync(policy, MY_UID, SERVICE1)) - .isEqualTo(Code.INTERNAL); + .isEqualTo(Status.INTERNAL.getCode()); assertThat(checkAuthorizationForServiceAsync(policy, OTHER_UID, SERVICE1)) - .isEqualTo(Code.INTERNAL); + .isEqualTo(Status.INTERNAL.getCode()); // Uses the specified policy for service2. assertThat(checkAuthorizationForServiceAsync(policy, MY_UID, SERVICE2)) - .isEqualTo(Code.PERMISSION_DENIED); - assertThat(checkAuthorizationForServiceAsync(policy, OTHER_UID, SERVICE2)).isEqualTo(Code.OK); + .isEqualTo(Status.PERMISSION_DENIED.getCode()); + assertThat(checkAuthorizationForServiceAsync(policy, OTHER_UID, SERVICE2)).isEqualTo(Status.OK.getCode()); // Falls back to the default. - assertThat(checkAuthorizationForServiceAsync(policy, MY_UID, SERVICE3)).isEqualTo(Code.OK); + assertThat(checkAuthorizationForServiceAsync(policy, MY_UID, SERVICE3)).isEqualTo(Status.OK.getCode()); assertThat(checkAuthorizationForServiceAsync(policy, OTHER_UID, SERVICE3)) - .isEqualTo(Code.PERMISSION_DENIED); + .isEqualTo(Status.PERMISSION_DENIED.getCode()); } @Test public void testPerServiceNoDefault_legacyApi() { @@ -246,58 +261,64 @@ SERVICE2, policy((uid) -> uid == OTHER_UID ? Status.OK : Status.PERMISSION_DENIE // Uses the specified policy for service1. assertThat(policy.checkAuthorizationForService(MY_UID, SERVICE1).getCode()) - .isEqualTo(Code.INTERNAL); + .isEqualTo(Status.INTERNAL.getCode()); assertThat(policy.checkAuthorizationForService(OTHER_UID, SERVICE1).getCode()) - .isEqualTo(Code.INTERNAL); + .isEqualTo(Status.INTERNAL.getCode()); // Uses the specified policy for service2. assertThat(policy.checkAuthorizationForService(MY_UID, SERVICE2).getCode()) - .isEqualTo(Code.PERMISSION_DENIED); + .isEqualTo(Status.PERMISSION_DENIED.getCode()); assertThat(policy.checkAuthorizationForService(OTHER_UID, SERVICE2).getCode()) - .isEqualTo(Code.OK); + .isEqualTo(Status.OK.getCode()); // Falls back to the default. assertThat(policy.checkAuthorizationForService(MY_UID, SERVICE3).getCode()) - .isEqualTo(Code.OK); + .isEqualTo(Status.OK.getCode()); assertThat(policy.checkAuthorizationForService(OTHER_UID, SERVICE3).getCode()) - .isEqualTo(Code.PERMISSION_DENIED); + .isEqualTo(Status.PERMISSION_DENIED.getCode()); } @Test public void testPerServiceNoDefaultAsync() throws Exception { policy = - ServerSecurityPolicy.newBuilder() - .servicePolicy( - SERVICE1, - asyncPolicy((uid) -> Futures.immediateFuture(Status.INTERNAL))) - .servicePolicy( - SERVICE2, asyncPolicy((uid) -> { - // Add some extra future transformation to confirm that a chain - // of futures gets properly handled. - ListenableFuture anotherUidFuture = - Futures.immediateFuture(uid == OTHER_UID); - return Futures - .transform( - anotherUidFuture, - anotherUid -> anotherUid ? Status.OK : Status.PERMISSION_DENIED, - MoreExecutors.directExecutor()); - })) - .build(); + ServerSecurityPolicy.newBuilder() + .servicePolicy( + SERVICE1, + asyncPolicy((uid) -> Futures.immediateFuture(Status.INTERNAL))) + .servicePolicy( + SERVICE2, asyncPolicy((uid) -> { + // Add some extra future transformation to confirm that a chain + // of futures gets properly handled. + ListenableFuture anotherUidFuture = + Futures.immediateFuture(uid == OTHER_UID); + return Futures + .transform( + anotherUidFuture, + anotherUid -> + anotherUid + ? Status.OK + : Status.PERMISSION_DENIED, + MoreExecutors.directExecutor()); + })) + .build(); // Uses the specified policy for service1. - assertThat(checkAuthorizationForServiceAsync(policy, MY_UID, SERVICE1)).isEqualTo(Code.INTERNAL); + assertThat(checkAuthorizationForServiceAsync(policy, MY_UID, SERVICE1)) + .isEqualTo(Status.INTERNAL.getCode()); assertThat(checkAuthorizationForServiceAsync(policy, OTHER_UID, SERVICE1)) - .isEqualTo(Code.INTERNAL); + .isEqualTo(Status.INTERNAL.getCode()); // Uses the specified policy for service2. assertThat(checkAuthorizationForServiceAsync(policy, MY_UID, SERVICE2)) - .isEqualTo(Code.PERMISSION_DENIED); - assertThat(checkAuthorizationForServiceAsync(policy, OTHER_UID, SERVICE2)).isEqualTo(Code.OK); + .isEqualTo(Status.PERMISSION_DENIED.getCode()); + assertThat(checkAuthorizationForServiceAsync(policy, OTHER_UID, SERVICE2)) + .isEqualTo(Status.OK.getCode()); // Falls back to the default. - assertThat(checkAuthorizationForServiceAsync(policy, MY_UID, SERVICE3)).isEqualTo(Code.OK); + assertThat(checkAuthorizationForServiceAsync(policy, MY_UID, SERVICE3)) + .isEqualTo(Status.OK.getCode()); assertThat(checkAuthorizationForServiceAsync(policy, OTHER_UID, SERVICE3)) - .isEqualTo(Code.PERMISSION_DENIED); + .isEqualTo(Status.PERMISSION_DENIED.getCode()); } /** @@ -333,12 +354,12 @@ public ListenableFuture checkAuthorizationAsync(int uid) { private static void awaitOrFail(CyclicBarrier barrier) { try { - barrier.await(); + barrier.await(); } catch (BrokenBarrierException e) { - fail(e.getMessage()); + fail(e.getMessage()); } catch (InterruptedException e) { - Thread.currentThread().interrupt(); - fail(e.getMessage()); + Thread.currentThread().interrupt(); + fail(e.getMessage()); } } } From 6f6cde9b1372c480bd730c470e615ec73a0b82da Mon Sep 17 00:00:00 2001 From: Mateus Azis Date: Thu, 30 Nov 2023 09:21:05 -0800 Subject: [PATCH 11/11] extra lint and style fixes --- .../io/grpc/binder/internal/BinderServer.java | 14 ++-- .../internal/BinderTransportSecurity.java | 65 +++++++++---------- 2 files changed, 37 insertions(+), 42 deletions(-) diff --git a/binder/src/main/java/io/grpc/binder/internal/BinderServer.java b/binder/src/main/java/io/grpc/binder/internal/BinderServer.java index 43a6dfcae47..8d6fe08fa94 100644 --- a/binder/src/main/java/io/grpc/binder/internal/BinderServer.java +++ b/binder/src/main/java/io/grpc/binder/internal/BinderServer.java @@ -35,8 +35,6 @@ import io.grpc.internal.ObjectPool; import io.grpc.internal.ServerListener; import io.grpc.internal.SharedResourceHolder; - -import java.io.Closeable; import java.io.IOException; import java.net.SocketAddress; import java.util.List; @@ -62,7 +60,7 @@ public final class BinderServer implements InternalServer, LeakSafeOneWayBinder. private final LeakSafeOneWayBinder hostServiceBinder; private final BinderTransportSecurity.ServerPolicyChecker serverPolicyChecker; private final InboundParcelablePolicy inboundParcelablePolicy; - private final BinderTransportSecurity.ShutdownListener shutdownListener; + private final BinderTransportSecurity.ShutdownListener transportSecurityShutdownListener; @GuardedBy("this") private ServerListener listener; @@ -74,8 +72,8 @@ public final class BinderServer implements InternalServer, LeakSafeOneWayBinder. private boolean shutdown; /** - * @param shutdownListener represents resources that should be cleaned up once the server shuts - * down. + * @param transportSecurityShutdownListener represents resources that should be cleaned up once + * the server shuts down. */ public BinderServer( AndroidComponentAddress listenAddress, @@ -83,14 +81,14 @@ public BinderServer( List streamTracerFactories, BinderTransportSecurity.ServerPolicyChecker serverPolicyChecker, InboundParcelablePolicy inboundParcelablePolicy, - BinderTransportSecurity.ShutdownListener shutdownListener) { + BinderTransportSecurity.ShutdownListener transportSecurityShutdownListener) { this.listenAddress = listenAddress; this.executorServicePool = executorServicePool; this.streamTracerFactories = ImmutableList.copyOf(checkNotNull(streamTracerFactories, "streamTracerFactories")); this.serverPolicyChecker = checkNotNull(serverPolicyChecker, "serverPolicyChecker"); this.inboundParcelablePolicy = inboundParcelablePolicy; - this.shutdownListener = shutdownListener; + this.transportSecurityShutdownListener = transportSecurityShutdownListener; hostServiceBinder = new LeakSafeOneWayBinder(this); } @@ -134,7 +132,7 @@ public synchronized void shutdown() { hostServiceBinder.detach(); listener.serverShutdown(); executorService = executorServicePool.returnObject(executorService); - shutdownListener.onShutdown(); + transportSecurityShutdownListener.onServerShutdown(); } } 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 c5e36ca9765..1866bf54a47 100644 --- a/binder/src/main/java/io/grpc/binder/internal/BinderTransportSecurity.java +++ b/binder/src/main/java/io/grpc/binder/internal/BinderTransportSecurity.java @@ -92,39 +92,6 @@ private static final class ServerAuthInterceptor implements ServerInterceptor { this.executor = executor; } - /** - * @param authStatusFuture a Future that is known to be complete, i.e. - * {@link ListenableFuture#isDone()} returns true. - */ - private ServerCall.Listener newServerCallListenerForPendingAuthResult( - ListenableFuture authStatusFuture, - ServerCall call, - Metadata headers, - ServerCallHandler next) { - PendingAuthListener listener = new PendingAuthListener<>(); - Futures.addCallback( - authStatusFuture, - new FutureCallback() { - @Override - public void onSuccess(Status authStatus) { - if (!authStatus.isOk()) { - call.close(authStatus, new Metadata()); - return; - } - - listener.startCall(call, headers, next); - } - - @Override - public void onFailure(Throwable t) { - call.close( - Status.INTERNAL.withCause(t).withDescription("Authorization future failed"), - new Metadata()); - } - }, executor); - return listener; - } - @Override public ServerCall.Listener interceptCall( ServerCall call, Metadata headers, ServerCallHandler next) { @@ -144,6 +111,7 @@ public ServerCall.Listener interceptCall( try { authStatus = Futures.getDone(authStatusFuture); } catch (ExecutionException e) { + // Failed futures are treated as an internal error rather than a security rejection. authStatus = Status.INTERNAL.withCause(e); } @@ -154,6 +122,35 @@ public ServerCall.Listener interceptCall( return new ServerCall.Listener() {}; } } + + private ServerCall.Listener newServerCallListenerForPendingAuthResult( + ListenableFuture authStatusFuture, + ServerCall call, + Metadata headers, + ServerCallHandler next) { + PendingAuthListener listener = new PendingAuthListener<>(); + Futures.addCallback( + authStatusFuture, + new FutureCallback() { + @Override + public void onSuccess(Status authStatus) { + if (!authStatus.isOk()) { + call.close(authStatus, new Metadata()); + return; + } + + listener.startCall(call, headers, next); + } + + @Override + public void onFailure(Throwable t) { + call.close( + Status.INTERNAL.withCause(t).withDescription("Authorization future failed"), + new Metadata()); + } + }, executor); + return listener; + } } /** @@ -230,6 +227,6 @@ public interface ServerPolicyChecker { * resources to be potentially cleaned up. */ public interface ShutdownListener { - void onShutdown(); + void onServerShutdown(); } }