binder: Cancel checkAuthorization() request if still pending upon termination

binder/src/androidTest/java/io/grpc/binder/internal/BinderClientTransportTest.java

@@ -17,15 +17,14 @@
 package io.grpc.binder.internal;
 
 import static com.google.common.truth.Truth.assertThat;
+import static java.util.concurrent.TimeUnit.SECONDS;
 
 import android.content.Context;
 import android.os.DeadObjectException;
 import android.os.Parcel;
 import android.os.RemoteException;
 import androidx.test.core.app.ApplicationProvider;
 import androidx.test.ext.junit.runners.AndroidJUnit4;
-import com.google.common.util.concurrent.Futures;
-import com.google.common.util.concurrent.ListenableFuture;
 import com.google.common.util.concurrent.SettableFuture;
 import com.google.errorprone.annotations.concurrent.GuardedBy;
 import com.google.protobuf.Empty;
@@ -38,13 +37,13 @@
 import io.grpc.Status;
 import io.grpc.Status.Code;
 import io.grpc.binder.AndroidComponentAddress;
-import io.grpc.binder.AsyncSecurityPolicy;
 import io.grpc.binder.BinderServerBuilder;
 import io.grpc.binder.HostServices;
 import io.grpc.binder.SecurityPolicy;
 import io.grpc.binder.internal.OneWayBinderProxies.BlackHoleOneWayBinderProxy;
 import io.grpc.binder.internal.OneWayBinderProxies.BlockingBinderDecorator;
 import io.grpc.binder.internal.OneWayBinderProxies.ThrowingOneWayBinderProxy;
+import io.grpc.binder.internal.SettableAsyncSecurityPolicy.AuthRequest;
 import io.grpc.internal.ClientStream;
 import io.grpc.internal.ClientStreamListener;
 import io.grpc.internal.ClientTransportFactory.ClientTransportOptions;
@@ -63,7 +62,6 @@
 import java.util.concurrent.Executors;
 import java.util.concurrent.LinkedBlockingQueue;
 import java.util.concurrent.ScheduledExecutorService;
-import java.util.concurrent.TimeUnit;
 import javax.annotation.Nullable;
 import org.junit.After;
 import org.junit.Before;
@@ -189,7 +187,7 @@ public void tearDown() throws Exception {
   private static void shutdownAndTerminate(ExecutorService executorService)
       throws InterruptedException {
     executorService.shutdownNow();
-    if (!executorService.awaitTermination(TIMEOUT_SECONDS, TimeUnit.SECONDS)) {
+    if (!executorService.awaitTermination(TIMEOUT_SECONDS, SECONDS)) {
       throw new AssertionError("executor failed to terminate promptly");
     }
   }
@@ -371,26 +369,32 @@ public void testBlackHoleEndpointConnectTimeout() throws Exception {
 
   @Test
   public void testBlackHoleSecurityPolicyConnectTimeout() throws Exception {
+    SettableAsyncSecurityPolicy securityPolicy = new SettableAsyncSecurityPolicy();
     transport =
         new BinderClientTransportBuilder()
-            .setSecurityPolicy(blockingSecurityPolicy)
+            .setSecurityPolicy(securityPolicy)
             .setReadyTimeoutMillis(1_234)
             .build();
     transport.start(transportListener).run();
+    // Take the next authRequest but don't respond to it, in order to trigger the ready timeout.
+    AuthRequest authRequest = securityPolicy.takeNextAuthRequest(TIMEOUT_SECONDS, SECONDS);
+
     Status transportStatus = transportListener.awaitShutdown();
     assertThat(transportStatus.getCode()).isEqualTo(Code.DEADLINE_EXCEEDED);
     assertThat(transportStatus.getDescription()).contains("1234");
     transportListener.awaitTermination();
-    blockingSecurityPolicy.provideNextCheckAuthorizationResult(Status.OK);
+
+    // If the transport gave up waiting on auth, it should cancel its request.
+    assertThat(authRequest.isCancelled()).isTrue();
   }
 
   @Test
   public void testAsyncSecurityPolicyFailure() throws Exception {
     SettableAsyncSecurityPolicy securityPolicy = new SettableAsyncSecurityPolicy();
     transport = new BinderClientTransportBuilder().setSecurityPolicy(securityPolicy).build();
     RuntimeException exception = new NullPointerException();
-    securityPolicy.setAuthorizationException(exception);
     transport.start(transportListener).run();
+    securityPolicy.takeNextAuthRequest(TIMEOUT_SECONDS, SECONDS).setResult(exception);
     Status transportStatus = transportListener.awaitShutdown();
     assertThat(transportStatus.getCode()).isEqualTo(Code.INTERNAL);
     assertThat(transportStatus.getCause()).isEqualTo(exception);
@@ -401,13 +405,27 @@ public void testAsyncSecurityPolicyFailure() throws Exception {
   public void testAsyncSecurityPolicySuccess() throws Exception {
     SettableAsyncSecurityPolicy securityPolicy = new SettableAsyncSecurityPolicy();
     transport = new BinderClientTransportBuilder().setSecurityPolicy(securityPolicy).build();
-    securityPolicy.setAuthorizationResult(Status.PERMISSION_DENIED);
     transport.start(transportListener).run();
+    securityPolicy
+        .takeNextAuthRequest(TIMEOUT_SECONDS, SECONDS)
+        .setResult(Status.PERMISSION_DENIED);
     Status transportStatus = transportListener.awaitShutdown();
     assertThat(transportStatus.getCode()).isEqualTo(Code.PERMISSION_DENIED);
     transportListener.awaitTermination();
   }
 
+  @Test
+  public void testAsyncSecurityPolicyCancelledUponExternalTermination() throws Exception {
+    SettableAsyncSecurityPolicy securityPolicy = new SettableAsyncSecurityPolicy();
+    transport = new BinderClientTransportBuilder().setSecurityPolicy(securityPolicy).build();
+    transport.start(transportListener).run();
+    AuthRequest authRequest = securityPolicy.takeNextAuthRequest(TIMEOUT_SECONDS, SECONDS);
+    transport.shutdownNow(Status.UNAVAILABLE); // 'authRequest' remains unanswered!
+    transportListener.awaitShutdown();
+    transportListener.awaitTermination();
+    assertThat(authRequest.isCancelled()).isTrue();
+  }
+
   private static void startAndAwaitReady(
       BinderTransport.BinderClientTransport transport, TestTransportListener transportListener)
       throws Exception {
@@ -429,7 +447,7 @@ public void transportShutdown(Status shutdownStatus) {
     }
 
     public Status awaitShutdown() throws Exception {
-      return shutdownStatus.get(TIMEOUT_SECONDS, TimeUnit.SECONDS);
+      return shutdownStatus.get(TIMEOUT_SECONDS, SECONDS);
     }
 
     @Override
@@ -440,7 +458,7 @@ public void transportTerminated() {
     }
 
     public void awaitTermination() throws Exception {
-      isTerminated.get(TIMEOUT_SECONDS, TimeUnit.SECONDS);
+      isTerminated.get(TIMEOUT_SECONDS, SECONDS);
     }
 
     @Override
@@ -451,7 +469,7 @@ public void transportReady() {
     }
 
     public void awaitReady() throws Exception {
-      isReady.get(TIMEOUT_SECONDS, TimeUnit.SECONDS);
+      isReady.get(TIMEOUT_SECONDS, SECONDS);
     }
 
     @Override
@@ -567,25 +585,4 @@ public Status checkAuthorization(int uid) {
       }
     }
   }
-
-  /** An AsyncSecurityPolicy that lets a test specify the outcome of checkAuthorizationAsync(). */
-  static class SettableAsyncSecurityPolicy extends AsyncSecurityPolicy {
-    private SettableFuture<Status> result = SettableFuture.create();
-
-    public void clearAuthorizationResult() {
-      result = SettableFuture.create();
-    }
-
-    public boolean setAuthorizationResult(Status status) {
-      return result.set(status);
-    }
-
-    public boolean setAuthorizationException(Throwable t) {
-      return result.setException(t);
-    }
-
-    public ListenableFuture<Status> checkAuthorizationAsync(int uid) {
-      return Futures.nonCancellationPropagating(result);
-    }
-  }
 }

binder/src/main/java/io/grpc/binder/internal/BinderTransport.java

@@ -577,6 +577,8 @@ public static final class BinderClientTransport extends BinderTransport
 
     @GuardedBy("this")
     private ScheduledFuture<?> readyTimeoutFuture; // != null iff timeout scheduled.
+    @GuardedBy("this")
+    @Nullable private ListenableFuture<Status> authResultFuture; // null before we check auth.
 
     /**
      * Constructs a new transport instance.
@@ -751,6 +753,9 @@ void notifyTerminated() {
         readyTimeoutFuture.cancel(false);
         readyTimeoutFuture = null;
       }
+      if (authResultFuture != null) {
+        authResultFuture.cancel(false);  // No effect if already complete.
+      }
       serviceBinding.unbind();
       clientTransportListener.transportTerminated();
     }
@@ -770,13 +775,13 @@ protected void handleSetupTransport(Parcel parcel) {
           shutdownInternal(
               Status.UNAVAILABLE.withDescription("Malformed SETUP_TRANSPORT data"), true);
         } else {
-          ListenableFuture<Status> authFuture =
+          authResultFuture =
               (securityPolicy instanceof AsyncSecurityPolicy)
                   ? ((AsyncSecurityPolicy) securityPolicy).checkAuthorizationAsync(remoteUid)
                   : Futures.submit(
                       () -> securityPolicy.checkAuthorization(remoteUid), offloadExecutor);
           Futures.addCallback(
-              authFuture,
+              authResultFuture,
               new FutureCallback<Status>() {
                 @Override
                 public void onSuccess(Status result) {

binder/src/testFixtures/java/io/grpc/binder/internal/SettableAsyncSecurityPolicy.java

@@ -0,0 +1,83 @@
+/*
+ * Copyright 2025 The gRPC Authors
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package io.grpc.binder.internal;
+
+import static com.google.common.base.Preconditions.checkState;
+
+import com.google.common.util.concurrent.ListenableFuture;
+import com.google.common.util.concurrent.SettableFuture;
+import io.grpc.Status;
+import io.grpc.binder.AsyncSecurityPolicy;
+import java.util.concurrent.LinkedBlockingDeque;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
+
+/**
+ * An {@link AsyncSecurityPolicy} that lets unit tests verify the exact order of authorization
+ * requests and respond to them one at a time.
+ */
+public class SettableAsyncSecurityPolicy extends AsyncSecurityPolicy {
+  private final LinkedBlockingDeque<AuthRequest> pendingRequests = new LinkedBlockingDeque<>();
+
+  @Override
+  public ListenableFuture<Status> checkAuthorizationAsync(int uid) {
+    AuthRequest request = new AuthRequest(uid);
+    pendingRequests.add(request);
+    return request.resultFuture;
+  }
+
+  /**
+   * Waits for the next "check authorization" request to be made and returns it, throwing in case no
+   * request arrives in time.
+   */
+  public AuthRequest takeNextAuthRequest(long timeout, TimeUnit unit)
+      throws InterruptedException, TimeoutException {
+    AuthRequest nextAuthRequest = pendingRequests.poll(timeout, unit);
+    if (nextAuthRequest == null) {
+      throw new TimeoutException();
+    }
+    return nextAuthRequest;
+  }
+
+  /** Represents a single call to {@link AsyncSecurityPolicy#checkAuthorizationAsync(int)}. */
+  public static class AuthRequest {
+
+    /** The argument passed to {@link AsyncSecurityPolicy#checkAuthorizationAsync(int)}. */
+    public final int uid;
+
+    private final SettableFuture<Status> resultFuture = SettableFuture.create();
+
+    private AuthRequest(int uid) {
+      this.uid = uid;
+    }
+
+    /** Provides this SecurityPolicy's response to this authorization request. */
+    public void setResult(Status result) {
+      checkState(resultFuture.set(result));
+    }
+
+    /** Simulates an exceptional response to this authorization request. */
+    public void setResult(Throwable t) {
+      checkState(resultFuture.setException(t));
+    }
+
+    /** Tests if the future returned for this authorization request was cancelled by the caller. */
+    public boolean isCancelled() {
+      return resultFuture.isCancelled();
+    }
+  }
+}