feat(mtls): Add support for X.509-based mTLS-transport in Java GAX lib (#3852)

Fixes #3851

- Refactors mTLS code path to use new CertificateBasedAccess class to
determine mTLS behavior based on env vars.
- Refactors mTLS code path to use DefaultMtlsProviderFactory from Java
auth lib for creating a default mTLS provider using either the legacy
SecureConnect mtls provider or the newer X.509 mtls provider, depending
on availability.

Author: andyrzhao

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

@@ -42,9 +42,12 @@
 import com.google.api.gax.rpc.TransportChannel;
 import com.google.api.gax.rpc.TransportChannelProvider;
 import com.google.api.gax.rpc.internal.EnvironmentProvider;
-import com.google.api.gax.rpc.mtls.MtlsProvider;
+import com.google.api.gax.rpc.mtls.CertificateBasedAccess;
 import com.google.auth.ApiKeyCredentials;
 import com.google.auth.Credentials;
+import com.google.auth.mtls.CertificateSourceUnavailableException;
+import com.google.auth.mtls.DefaultMtlsProviderFactory;
+import com.google.auth.mtls.MtlsProvider;
 import com.google.auth.oauth2.ComputeEngineCredentials;
 import com.google.auth.oauth2.SecureSessionAgent;
 import com.google.auth.oauth2.SecureSessionAgentConfig;
@@ -150,6 +153,7 @@ public final class InstantiatingGrpcChannelProvider implements TransportChannelP
   @Nullable private final Boolean allowNonDefaultServiceAccount;
   @VisibleForTesting final ImmutableMap<String, ?> directPathServiceConfig;
   @Nullable private final MtlsProvider mtlsProvider;
+  private final CertificateBasedAccess certificateBasedAccess;
   @Nullable private final SecureSessionAgent s2aConfigProvider;
   private final List<HardBoundTokenTypes> allowedHardBoundTokenTypes;
   @VisibleForTesting final Map<String, String> headersWithDuplicatesRemoved = new HashMap<>();
@@ -183,6 +187,7 @@ private InstantiatingGrpcChannelProvider(Builder builder) {
     this.mtlsEndpoint = builder.mtlsEndpoint;
     this.allowedHardBoundTokenTypes = builder.allowedHardBoundTokenTypes;
     this.mtlsProvider = builder.mtlsProvider;
+    this.certificateBasedAccess = builder.certificateBasedAccess;
     this.s2aConfigProvider = builder.s2aConfigProvider;
     this.envProvider = builder.envProvider;
     this.interceptorProvider = builder.interceptorProvider;
@@ -484,7 +489,10 @@ boolean canUseDirectPathWithUniverseDomain() {
 
   @VisibleForTesting
   ChannelCredentials createMtlsChannelCredentials() throws IOException, GeneralSecurityException {
-    if (mtlsProvider.useMtlsClientCertificate()) {
+    if (mtlsProvider == null) {
+      return null;
+    }
+    if (certificateBasedAccess.useMtlsClientCertificate()) {
       KeyStore mtlsKeyStore = mtlsProvider.getKeyStore();
       if (mtlsKeyStore != null) {
         KeyManagerFactory factory =
@@ -853,7 +861,8 @@ public static final class Builder {
     private boolean useS2A;
     private EnvironmentProvider envProvider;
     private SecureSessionAgent s2aConfigProvider = SecureSessionAgent.create();
-    private MtlsProvider mtlsProvider = new MtlsProvider();
+    @Nullable private MtlsProvider mtlsProvider;
+    private CertificateBasedAccess certificateBasedAccess;
     @Nullable private GrpcInterceptorProvider interceptorProvider;
     @Nullable private Integer maxInboundMessageSize;
     @Nullable private Integer maxInboundMetadataSize;
@@ -904,6 +913,7 @@ private Builder(InstantiatingGrpcChannelProvider provider) {
       this.allowedHardBoundTokenTypes = provider.allowedHardBoundTokenTypes;
       this.directPathServiceConfig = provider.directPathServiceConfig;
       this.mtlsProvider = provider.mtlsProvider;
+      this.certificateBasedAccess = provider.certificateBasedAccess;
       this.s2aConfigProvider = provider.s2aConfigProvider;
     }
 
@@ -994,6 +1004,12 @@ Builder setMtlsProvider(MtlsProvider mtlsProvider) {
       return this;
     }
 
+    @VisibleForTesting
+    Builder setCertificateBasedAccess(CertificateBasedAccess certificateBasedAccess) {
+      this.certificateBasedAccess = certificateBasedAccess;
+      return this;
+    }
+
     @VisibleForTesting
     Builder setS2AConfigProvider(SecureSessionAgent s2aConfigProvider) {
       this.s2aConfigProvider = s2aConfigProvider;
@@ -1269,6 +1285,25 @@ CallCredentials createHardBoundTokensCallCredentials(
     }
 
     public InstantiatingGrpcChannelProvider build() {
+      if (certificateBasedAccess == null) {
+        certificateBasedAccess = CertificateBasedAccess.createWithSystemEnv();
+      }
+      if (certificateBasedAccess.useMtlsClientCertificate()) {
+        if (mtlsProvider == null) {
+          // Attempt to create default MtlsProvider from environment.
+          try {
+            mtlsProvider = DefaultMtlsProviderFactory.create();
+          } catch (CertificateSourceUnavailableException e) {
+            // This is okay. Leave mtlsProvider as null so that we will not auto-upgrade
+            // to mTLS endpoints. See https://google.aip.dev/auth/4114.
+          } catch (IOException e) {
+            LOG.log(
+                Level.WARNING,
+                "DefaultMtlsProviderFactory encountered unexpected IOException: " + e.getMessage());
+          }
+        }
+      }
+
       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.