feat: Enable MTLS_S2A bound token by default for gRPC S2A enabled flows

gax-java/gax-grpc/src/main/java/com/google/api/gax/grpc/InstantiatingGrpcChannelProvider.java

@@ -56,6 +56,7 @@
 import com.google.common.io.Files;
 import io.grpc.CallCredentials;
 import io.grpc.ChannelCredentials;
+import io.grpc.CompositeChannelCredentials;
 import io.grpc.Grpc;
 import io.grpc.InsecureChannelCredentials;
 import io.grpc.ManagedChannel;
@@ -592,6 +593,41 @@ ChannelCredentials createS2ASecuredChannelCredentials() {
     }
   }
 
+  boolean isMtlsS2AHardBoundTokensEnabled() {
+    if (!useS2A) {
+      // If S2A cannot be used, {@code HardBoundTokenTypes.MTLS_S2A} hard bound tokens should not be
+      // used
+      return false;
+    }
+    if (allowedHardBoundTokenTypes == null
+        || credentials == null
+        || !(credentials instanceof ComputeEngineCredentials)) {
+      return false;
+    }
+    for (HardBoundTokenTypes boundTokenTypes : allowedHardBoundTokenTypes) {
+      if (boundTokenTypes == HardBoundTokenTypes.MTLS_S2A) {
+        return true;
+      }
+    }
+    return false;
+  }
+
+  CallCredentials createHardBoundTokensCallCredentials(
+      ComputeEngineCredentials.GoogleAuthTransport googleAuthTransport,
+      ComputeEngineCredentials.BindingEnforcement bindingEnforcement) {
+    ComputeEngineCredentials.Builder credsBuilder =
+        ((ComputeEngineCredentials) credentials).toBuilder();
+    // We only set scopes and HTTP transport factory from the original credentials because
+    // only those are used in gRPC CallCredentials to fetch request metadata.
+    return MoreCallCredentials.from(
+        ComputeEngineCredentials.newBuilder()
+            .setScopes(credsBuilder.getScopes())
+            .setHttpTransportFactory(credsBuilder.getHttpTransportFactory())
+            .setGoogleAuthTransport(googleAuthTransport)
+            .setBindingEnforcement(bindingEnforcement)
+            .build());
+  }
+
   private ManagedChannel createSingleChannel() throws IOException {
     GrpcHeaderInterceptor headerInterceptor =
         new GrpcHeaderInterceptor(headersWithDuplicatesRemoved);
@@ -648,6 +684,15 @@ private ManagedChannel createSingleChannel() throws IOException {
         }
         if (channelCredentials != null) {
           // Create the channel using S2A-secured channel credentials.
+          if (isMtlsS2AHardBoundTokensEnabled()) {
+            // Set a {@code ComputeEngineCredentials} instance to be per-RPC call credentials,
+            // which will be used to fetch MTLS_S2A hard bound tokens from the metdata server.
+            CallCredentials callCreds =
+                createHardBoundTokensCallCredentials(
+                    ComputeEngineCredentials.GoogleAuthTransport.MTLS,
+                    ComputeEngineCredentials.BindingEnforcement.ON);
+            channelCredentials = CompositeChannelCredentials.create(channelCredentials, callCreds);
+          }
           builder = Grpc.newChannelBuilder(endpoint, channelCredentials);
         } else {
           // Use default if we cannot initialize channel credentials via DCA or S2A.

gax-java/gax-grpc/src/test/java/com/google/api/gax/grpc/InstantiatingGrpcChannelProviderTest.java

@@ -1103,6 +1103,82 @@ void createS2ASecuredChannelCredentials_returnsPlaintextToS2AS2AChannelCredentia
     InstantiatingGrpcChannelProvider.LOG.removeHandler(logHandler);
   }
 
+  @Test
+  void isMtlsS2AHardBoundTokensEnabled_useS2AFalse() {
+    InstantiatingGrpcChannelProvider provider =
+        InstantiatingGrpcChannelProvider.newBuilder()
+            .setUseS2A(false)
+            .setAllowHardBoundTokenTypes(
+                Collections.singletonList(
+                    InstantiatingGrpcChannelProvider.HardBoundTokenTypes.MTLS_S2A))
+            .setCredentials(computeEngineCredentials)
+            .build();
+    Truth.assertThat(provider.isMtlsS2AHardBoundTokensEnabled()).isFalse();
+  }
+
+  @Test
+  void isMtlsS2AHardBoundTokensEnabled_hardBoundTokenTypesNull() {
+    InstantiatingGrpcChannelProvider provider =
+        InstantiatingGrpcChannelProvider.newBuilder()
+            .setUseS2A(true)
+            .setAllowHardBoundTokenTypes(null)
+            .setCredentials(computeEngineCredentials)
+            .build();
+    Truth.assertThat(provider.isMtlsS2AHardBoundTokensEnabled()).isFalse();
+  }
+
+  @Test
+  void isMtlsS2AHardBoundTokensEnabled_nullCreds() {
+    InstantiatingGrpcChannelProvider provider =
+        InstantiatingGrpcChannelProvider.newBuilder()
+            .setUseS2A(true)
+            .setAllowHardBoundTokenTypes(
+                Collections.singletonList(
+                    InstantiatingGrpcChannelProvider.HardBoundTokenTypes.MTLS_S2A))
+            .setCredentials(null)
+            .build();
+    Truth.assertThat(provider.isMtlsS2AHardBoundTokensEnabled()).isFalse();
+  }
+
+  @Test
+  void isMtlsS2AHardBoundTokensEnabled_notComputeEngineCreds() {
+    InstantiatingGrpcChannelProvider provider =
+        InstantiatingGrpcChannelProvider.newBuilder()
+            .setUseS2A(true)
+            .setAllowHardBoundTokenTypes(
+                Collections.singletonList(
+                    InstantiatingGrpcChannelProvider.HardBoundTokenTypes.MTLS_S2A))
+            .setCredentials(CloudShellCredentials.create(3000))
+            .build();
+    Truth.assertThat(provider.isMtlsS2AHardBoundTokensEnabled()).isFalse();
+  }
+
+  @Test
+  void isMtlsS2AHardBoundTokensEnabled_mtlsS2ANotInList() {
+    InstantiatingGrpcChannelProvider provider =
+        InstantiatingGrpcChannelProvider.newBuilder()
+            .setUseS2A(true)
+            .setAllowHardBoundTokenTypes(
+                Collections.singletonList(
+                    InstantiatingGrpcChannelProvider.HardBoundTokenTypes.ALTS))
+            .setCredentials(computeEngineCredentials)
+            .build();
+    Truth.assertThat(provider.isMtlsS2AHardBoundTokensEnabled()).isFalse();
+  }
+
+  @Test
+  void isMtlsS2AHardBoundTokensEnabled_returnsTrue() {
+    InstantiatingGrpcChannelProvider provider =
+        InstantiatingGrpcChannelProvider.newBuilder()
+            .setUseS2A(true)
+            .setAllowHardBoundTokenTypes(
+                Collections.singletonList(
+                    InstantiatingGrpcChannelProvider.HardBoundTokenTypes.MTLS_S2A))
+            .setCredentials(computeEngineCredentials)
+            .build();
+    Truth.assertThat(provider.isMtlsS2AHardBoundTokensEnabled()).isTrue();
+  }
+
   private static class FakeLogHandler extends Handler {
 
     List<LogRecord> records = new ArrayList<>();