feat: revert #3400: reintroduce experimental S2A integration in client libraries grpc transport (#3548)

**Revert #3400.**

**This PR re-introduces the S2A integration the Java Cloud SDK
(initially introduced in #3326, and temporarily reverted in #3400).**

**This PR does this by reverting #3400 with the following patches:**
- load the S2A APIs via reflection. This allows us to merge the code
while the [S2A API is still experimental in
gRPC-Java](https://github.com/grpc/grpc-java/blob/master/s2a/src/main/java/io/grpc/s2a/S2AChannelCredentials.java)
without introducing a diamond dependency conflict. Once the S2A APIs are
stable, the reflection logic can be removed and the S2A API can be used
directly (via a dependency on S2A API)
- fix NPE (#3401)
- use a different env var name for enabling the feature


**Below is the original description from #3326**

Modify the Client Libraries gRPC Channel builder to use mTLS via S2A if
the experimental environment variable is set, S2A is available (We check
this by using [SecureSessionAgent
utility](https://github.com/googleapis/google-auth-library-java/blob/main/oauth2_http/java/com/google/auth/oauth2/SecureSessionAgent.java)),
and a few more conditions (see `shouldUseS2A`).

Following https://google.aip.dev/auth/4115, Only attempt to use S2A
after DirectPath and DCA (https://google.aip.dev/auth/4114) are ruled
out as options. If conditions to use S2A are not met (env variable not
set, or S2A is not running in environment, etc (`shouldUseS2A` returns
false)), fall back to default TLS connection.

When we are creating S2A-enabled Grpc Channel Credentials, we first try
to secure the connection between the client and the S2A via MTLS, using
[MTLS-MDS](https://cloud.google.com/compute/docs/metadata/overview#https-mds)
credentials. If MTLS-MDS credentials can't be loaded, then we fallback
to a plaintext connection between the client and S2A.

The parallel go implementation : googleapis/google-api-go-client#1874
(now lives here:
https://github.com/googleapis/google-cloud-go/blob/main/auth/internal/transport/cba.go)

S2A Java client: https://github.com/grpc/grpc-java/tree/master/s2a

Resolving b/376258193 means that S2A.java is no longer experimental

Author: rmehta19

gax-java/gax-grpc/clirr-ignored-differences.xml

@@ -7,10 +7,4 @@
     <className>com/google/api/gax/grpc/GrpcTransportChannel</className>
     <method>boolean isDirectPath()</method>
   </difference>
-  <!-- Ignore this as this was part of s2a-grpc ExperimentalApi revert -->
-  <difference>
-    <differenceType>7002</differenceType>
-    <className>com/google/api/gax/grpc/InstantiatingGrpcChannelProvider</className>
-    <method>* withUseS2A(*)</method>
-  </difference>
 </differences>

gax-java/gax-grpc/pom.xml

@@ -94,6 +94,11 @@
     </dependency>
 
     <!-- test dependencies -->
+    <dependency>
+      <groupId>io.grpc</groupId>
+      <artifactId>grpc-s2a</artifactId>
+      <scope>test</scope>
+    </dependency>
     <dependency>
       <groupId>com.google.api.grpc</groupId>
       <artifactId>grpc-google-common-protos</artifactId>

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

@@ -46,21 +46,26 @@
 import com.google.auth.ApiKeyCredentials;
 import com.google.auth.Credentials;
 import com.google.auth.oauth2.ComputeEngineCredentials;
+import com.google.auth.oauth2.SecureSessionAgent;
+import com.google.auth.oauth2.SecureSessionAgentConfig;
 import com.google.common.annotations.VisibleForTesting;
 import com.google.common.base.Preconditions;
+import com.google.common.base.Strings;
 import com.google.common.collect.ImmutableList;
 import com.google.common.collect.ImmutableMap;
 import com.google.common.io.Files;
 import io.grpc.CallCredentials;
 import io.grpc.ChannelCredentials;
 import io.grpc.Grpc;
+import io.grpc.InsecureChannelCredentials;
 import io.grpc.ManagedChannel;
 import io.grpc.ManagedChannelBuilder;
 import io.grpc.TlsChannelCredentials;
 import io.grpc.alts.GoogleDefaultChannelCredentials;
 import io.grpc.auth.MoreCallCredentials;
 import java.io.File;
 import java.io.IOException;
+import java.lang.reflect.Method;
 import java.nio.charset.StandardCharsets;
 import java.security.GeneralSecurityException;
 import java.security.KeyStore;
@@ -99,6 +104,19 @@ public final class InstantiatingGrpcChannelProvider implements TransportChannelP
   @VisibleForTesting
   static final String DIRECT_PATH_ENV_ENABLE_XDS = "GOOGLE_CLOUD_ENABLE_DIRECT_PATH_XDS";
 
+  // The public portion of the mTLS MDS root certificate is stored for performing
+  // cert verification when establishing an mTLS connection with the MDS. See
+  // {@link <a
+  // href="https://cloud.google.com/compute/docs/metadata/overview#https-mds-root-certs">this</a>
+  // for more information.}
+  private static final String MTLS_MDS_ROOT_PATH = "/run/google-mds-mtls/root.crt";
+  // The mTLS MDS credentials are formatted as the concatenation of a PEM-encoded certificate chain
+  // followed by a PEM-encoded private key. See
+  // {@link <a
+  // href="https://cloud.google.com/compute/docs/metadata/overview#https-mds-client-certs">this</a>
+  // for more information.}
+  private static final String MTLS_MDS_CERT_CHAIN_AND_KEY_PATH = "/run/google-mds-mtls/client.key";
+
   static final long DIRECT_PATH_KEEP_ALIVE_TIME_SECONDS = 3600;
   static final long DIRECT_PATH_KEEP_ALIVE_TIMEOUT_SECONDS = 20;
   static final String GCE_PRODUCTION_NAME_PRIOR_2016 = "Google";
@@ -107,6 +125,7 @@ public final class InstantiatingGrpcChannelProvider implements TransportChannelP
   private final int processorCount;
   private final Executor executor;
   private final HeaderProvider headerProvider;
+  private final boolean useS2A;
   private final String endpoint;
   // TODO: remove. envProvider currently provides DirectPath environment variable, and is only used
   // during initial rollout for DirectPath. This provider will be removed once the DirectPath
@@ -126,6 +145,7 @@ public final class InstantiatingGrpcChannelProvider implements TransportChannelP
   @Nullable private final Boolean allowNonDefaultServiceAccount;
   @VisibleForTesting final ImmutableMap<String, ?> directPathServiceConfig;
   @Nullable private final MtlsProvider mtlsProvider;
+  @Nullable private final SecureSessionAgent s2aConfigProvider;
   @Nullable private final List<HardBoundTokenTypes> allowedHardBoundTokenTypes;
   @VisibleForTesting final Map<String, String> headersWithDuplicatesRemoved = new HashMap<>();
 
@@ -153,9 +173,11 @@ private InstantiatingGrpcChannelProvider(Builder builder) {
     this.processorCount = builder.processorCount;
     this.executor = builder.executor;
     this.headerProvider = builder.headerProvider;
+    this.useS2A = builder.useS2A;
     this.endpoint = builder.endpoint;
     this.allowedHardBoundTokenTypes = builder.allowedHardBoundTokenTypes;
     this.mtlsProvider = builder.mtlsProvider;
+    this.s2aConfigProvider = builder.s2aConfigProvider;
     this.envProvider = builder.envProvider;
     this.interceptorProvider = builder.interceptorProvider;
     this.maxInboundMessageSize = builder.maxInboundMessageSize;
@@ -244,6 +266,17 @@ public TransportChannelProvider withEndpoint(String endpoint) {
     return toBuilder().setEndpoint(endpoint).build();
   }
 
+  /**
+   * Specify whether or not to use S2A.
+   *
+   * @param useS2A
+   * @return A new {@link InstantiatingGrpcChannelProvider} with useS2A set.
+   */
+  @Override
+  public TransportChannelProvider withUseS2A(boolean useS2A) {
+    return toBuilder().setUseS2A(useS2A).build();
+  }
+
   /** @deprecated Please modify pool settings via {@link #toBuilder()} */
   @Deprecated
   @Override
@@ -429,6 +462,136 @@ ChannelCredentials createMtlsChannelCredentials() throws IOException, GeneralSec
     return null;
   }
 
+  /**
+   * Create the S2A-Secured Channel credentials. Load the API using reflection. Once the S2A API is
+   * stable in gRPC-Java, all callers of this method can simply use the S2A APIs directly and this
+   * method can be deleted.
+   *
+   * @param s2aAddress the address of the S2A server used to secure the connection.
+   * @param s2aChannelCredentials the credentials to be used when connecting to the S2A.
+   * @return {@code ChannelCredentials} instance.
+   */
+  ChannelCredentials buildS2AChannelCredentials(
+      String s2aAddress, ChannelCredentials s2aChannelCredentials) {
+    try {
+      // Load the S2A API.
+      Class<?> s2aChannelCreds = Class.forName("io.grpc.s2a.S2AChannelCredentials");
+      Class<?> s2aChannelCredsBuilder = Class.forName("io.grpc.s2a.S2AChannelCredentials$Builder");
+
+      // Load and invoke the S2A API methods.
+      Class<?>[] partypes = new Class[2];
+      partypes[0] = String.class;
+      partypes[1] = ChannelCredentials.class;
+      Method newBuilder = s2aChannelCreds.getMethod("newBuilder", partypes);
+      Object arglist[] = new Object[2];
+      arglist[0] = s2aAddress;
+      arglist[1] = s2aChannelCredentials;
+      Object retObjBuilder = newBuilder.invoke(null, arglist);
+      Method build = s2aChannelCredsBuilder.getMethod("build", null);
+      Object retObjCreds = build.invoke(retObjBuilder, null);
+      return (ChannelCredentials) retObjCreds;
+    } catch (Throwable t) {
+      LOG.log(
+          Level.WARNING,
+          "Falling back to default (TLS without S2A) because S2A APIs cannot be used: "
+              + t.getMessage());
+      return null;
+    }
+  }
+
+  /**
+   * This method creates {@link TlsChannelCredentials} to be used by the client to establish an mTLS
+   * connection to S2A. Returns null if any of {@param trustBundle}, {@param privateKey} or {@param
+   * certChain} are missing.
+   *
+   * @param trustBundle the trust bundle to be used to establish the client -> S2A mTLS connection
+   * @param privateKey the client's private key to be used to establish the client -> S2A mtls
+   *     connection
+   * @param certChain the client's cert chain to be used to establish the client -> S2A mtls
+   *     connection
+   * @return {@link ChannelCredentials} to use to create an mtls connection between client and S2A
+   * @throws IOException on error
+   */
+  @VisibleForTesting
+  ChannelCredentials createMtlsToS2AChannelCredentials(
+      File trustBundle, File privateKey, File certChain) throws IOException {
+    if (trustBundle == null || privateKey == null || certChain == null) {
+      return null;
+    }
+    return TlsChannelCredentials.newBuilder()
+        .keyManager(privateKey, certChain)
+        .trustManager(trustBundle)
+        .build();
+  }
+
+  /**
+   * This method creates {@link ChannelCredentials} to be used by client to establish a plaintext
+   * connection to S2A. if {@param plaintextAddress} is not present, returns null.
+   *
+   * @param plaintextAddress the address to reach S2A which accepts plaintext connections
+   * @return {@link ChannelCredentials} to use to create a plaintext connection between client and
+   *     S2A
+   */
+  ChannelCredentials createPlaintextToS2AChannelCredentials(String plaintextAddress) {
+    if (Strings.isNullOrEmpty(plaintextAddress)) {
+      return null;
+    }
+    return buildS2AChannelCredentials(plaintextAddress, InsecureChannelCredentials.create());
+  }
+
+  /**
+   * This method creates gRPC {@link ChannelCredentials} configured to use S2A to estbalish a mTLS
+   * connection. First, the address of S2A is discovered by using the {@link S2A} utility to learn
+   * the {@code mtlsAddress} to reach S2A and the {@code plaintextAddress} to reach S2A. Prefer to
+   * use the {@code mtlsAddress} address to reach S2A if it is non-empty and the MTLS-MDS
+   * credentials can successfully be discovered and used to create {@link TlsChannelCredentials}. If
+   * there is any failure using mTLS-to-S2A, fallback to using a plaintext connection to S2A using
+   * the {@code plaintextAddress}. If {@code plaintextAddress} is not available, this function
+   * returns null; in this case S2A will not be used, and a TLS connection to the service will be
+   * established.
+   *
+   * @return {@link ChannelCredentials} configured to use S2A to create mTLS connection.
+   */
+  ChannelCredentials createS2ASecuredChannelCredentials() {
+    SecureSessionAgentConfig config = s2aConfigProvider.getConfig();
+    String plaintextAddress = config.getPlaintextAddress();
+    String mtlsAddress = config.getMtlsAddress();
+    if (Strings.isNullOrEmpty(mtlsAddress)) {
+      // Fallback to plaintext connection to S2A.
+      LOG.log(
+          Level.INFO,
+          "Cannot establish an mTLS connection to S2A because autoconfig endpoint did not return a mtls address to reach S2A.");
+      return createPlaintextToS2AChannelCredentials(plaintextAddress);
+    }
+    // Currently, MTLS to MDS is only available on GCE. See:
+    // https://cloud.google.com/compute/docs/metadata/overview#https-mds
+    // Try to load MTLS-MDS creds.
+    File rootFile = new File(MTLS_MDS_ROOT_PATH);
+    File certKeyFile = new File(MTLS_MDS_CERT_CHAIN_AND_KEY_PATH);
+    if (rootFile.isFile() && certKeyFile.isFile()) {
+      // Try to connect to S2A using mTLS.
+      ChannelCredentials mtlsToS2AChannelCredentials = null;
+      try {
+        mtlsToS2AChannelCredentials =
+            createMtlsToS2AChannelCredentials(rootFile, certKeyFile, certKeyFile);
+      } catch (IOException ignore) {
+        // Fallback to plaintext-to-S2A connection on error.
+        LOG.log(
+            Level.WARNING,
+            "Cannot establish an mTLS connection to S2A due to error creating MTLS to MDS TlsChannelCredentials credentials, falling back to plaintext connection to S2A: "
+                + ignore.getMessage());
+        return createPlaintextToS2AChannelCredentials(plaintextAddress);
+      }
+      return buildS2AChannelCredentials(mtlsAddress, mtlsToS2AChannelCredentials);
+    } else {
+      // Fallback to plaintext-to-S2A connection if MTLS-MDS creds do not exist.
+      LOG.log(
+          Level.INFO,
+          "Cannot establish an mTLS connection to S2A because MTLS to MDS credentials do not exist on filesystem, falling back to plaintext connection to S2A");
+      return createPlaintextToS2AChannelCredentials(plaintextAddress);
+    }
+  }
+
   private ManagedChannel createSingleChannel() throws IOException {
     GrpcHeaderInterceptor headerInterceptor =
         new GrpcHeaderInterceptor(headersWithDuplicatesRemoved);
@@ -468,14 +631,28 @@ private ManagedChannel createSingleChannel() throws IOException {
     } else {
       ChannelCredentials channelCredentials;
       try {
+        // Try and create credentials via DCA. See https://google.aip.dev/auth/4114.
         channelCredentials = createMtlsChannelCredentials();
       } catch (GeneralSecurityException e) {
         throw new IOException(e);
       }
       if (channelCredentials != null) {
+        // Create the channel using channel credentials created via DCA.
         builder = Grpc.newChannelBuilder(endpoint, channelCredentials);
       } else {
-        builder = ManagedChannelBuilder.forAddress(serviceAddress, port);
+        // Could not create channel credentials via DCA. In accordance with
+        // https://google.aip.dev/auth/4115, if credentials not available through
+        // DCA, try mTLS with credentials held by the S2A (Secure Session Agent).
+        if (useS2A) {
+          channelCredentials = createS2ASecuredChannelCredentials();
+        }
+        if (channelCredentials != null) {
+          // Create the channel using S2A-secured channel credentials.
+          builder = Grpc.newChannelBuilder(endpoint, channelCredentials);
+        } else {
+          // Use default if we cannot initialize channel credentials via DCA or S2A.
+          builder = ManagedChannelBuilder.forAddress(serviceAddress, port);
+        }
       }
     }
     // google-c2p resolver requires service config lookup
@@ -623,7 +800,9 @@ public static final class Builder {
     private Executor executor;
     private HeaderProvider headerProvider;
     private String endpoint;
+    private boolean useS2A;
     private EnvironmentProvider envProvider;
+    private SecureSessionAgent s2aConfigProvider = SecureSessionAgent.create();
     private MtlsProvider mtlsProvider = new MtlsProvider();
     @Nullable private GrpcInterceptorProvider interceptorProvider;
     @Nullable private Integer maxInboundMessageSize;
@@ -652,6 +831,7 @@ private Builder(InstantiatingGrpcChannelProvider provider) {
       this.executor = provider.executor;
       this.headerProvider = provider.headerProvider;
       this.endpoint = provider.endpoint;
+      this.useS2A = provider.useS2A;
       this.envProvider = provider.envProvider;
       this.interceptorProvider = provider.interceptorProvider;
       this.maxInboundMessageSize = provider.maxInboundMessageSize;
@@ -668,6 +848,7 @@ private Builder(InstantiatingGrpcChannelProvider provider) {
       this.allowNonDefaultServiceAccount = provider.allowNonDefaultServiceAccount;
       this.directPathServiceConfig = provider.directPathServiceConfig;
       this.mtlsProvider = provider.mtlsProvider;
+      this.s2aConfigProvider = provider.s2aConfigProvider;
     }
 
     /**
@@ -720,6 +901,10 @@ public Builder setEndpoint(String endpoint) {
       return this;
     }
 
+    Builder setUseS2A(boolean useS2A) {
+      this.useS2A = useS2A;
+      return this;
+    }
     /*
      * Sets the allowed hard bound token types for this TransportChannelProvider.
      *
@@ -739,6 +924,12 @@ Builder setMtlsProvider(MtlsProvider mtlsProvider) {
       return this;
     }
 
+    @VisibleForTesting
+    Builder setS2AConfigProvider(SecureSessionAgent s2aConfigProvider) {
+      this.s2aConfigProvider = s2aConfigProvider;
+      return this;
+    }
+
     /**
      * Sets the GrpcInterceptorProvider for this TransportChannelProvider.
      *

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

@@ -101,6 +101,8 @@ void setUp() throws IOException {
     TransportChannel transportChannel =
         GrpcTransportChannel.newBuilder().setManagedChannel(channel).build();
     when(operationsChannelProvider.getTransportChannel()).thenReturn(transportChannel);
+    when(operationsChannelProvider.withUseS2A(Mockito.any(boolean.class)))
+        .thenReturn(operationsChannelProvider);
 
     clock = new FakeApiClock(0L);
     executor = RecordingScheduler.create(clock);

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

@@ -51,12 +51,16 @@
 import com.google.auth.http.AuthHttpConstants;
 import com.google.auth.oauth2.CloudShellCredentials;
 import com.google.auth.oauth2.ComputeEngineCredentials;
+import com.google.auth.oauth2.SecureSessionAgent;
+import com.google.auth.oauth2.SecureSessionAgentConfig;
 import com.google.common.collect.ImmutableList;
 import com.google.common.collect.ImmutableMap;
 import com.google.common.truth.Truth;
 import io.grpc.ManagedChannel;
 import io.grpc.ManagedChannelBuilder;
+import io.grpc.TlsChannelCredentials;
 import io.grpc.alts.ComputeEngineChannelBuilder;
+import java.io.File;
 import java.io.IOException;
 import java.security.GeneralSecurityException;
 import java.time.Duration;
@@ -985,6 +989,120 @@ private FixedHeaderProvider getHeaderProviderWithApiKeyHeader() {
     return FixedHeaderProvider.create(header);
   }
 
+  @Test
+  void createPlaintextToS2AChannelCredentials_emptyPlaintextAddress_returnsNull() {
+    InstantiatingGrpcChannelProvider provider =
+        InstantiatingGrpcChannelProvider.newBuilder().build();
+    assertThat(provider.createPlaintextToS2AChannelCredentials("")).isNull();
+  }
+
+  @Test
+  void createPlaintextToS2AChannelCredentials_success() {
+    InstantiatingGrpcChannelProvider provider =
+        InstantiatingGrpcChannelProvider.newBuilder().build();
+    assertThat(provider.createPlaintextToS2AChannelCredentials("localhost:8080")).isNotNull();
+  }
+
+  @Test
+  void createMtlsToS2AChannelCredentials_missingAllFiles_throws() throws IOException {
+    InstantiatingGrpcChannelProvider provider =
+        InstantiatingGrpcChannelProvider.newBuilder().build();
+    assertThat(provider.createMtlsToS2AChannelCredentials(null, null, null)).isNull();
+  }
+
+  @Test
+  void createMtlsToS2AChannelCredentials_missingRootFile_throws() throws IOException {
+    InstantiatingGrpcChannelProvider provider =
+        InstantiatingGrpcChannelProvider.newBuilder().build();
+    File privateKey = new File("src/test/resources/client_key.pem");
+    File certChain = new File("src/test/resources/client_cert.pem");
+    assertThat(provider.createMtlsToS2AChannelCredentials(null, privateKey, certChain)).isNull();
+  }
+
+  @Test
+  void createMtlsToS2AChannelCredentials_missingKeyFile_throws() throws IOException {
+    InstantiatingGrpcChannelProvider provider =
+        InstantiatingGrpcChannelProvider.newBuilder().build();
+    File trustBundle = new File("src/test/resources/root_cert.pem");
+    File certChain = new File("src/test/resources/client_cert.pem");
+    assertThat(provider.createMtlsToS2AChannelCredentials(trustBundle, null, certChain)).isNull();
+  }
+
+  @Test
+  void createMtlsToS2AChannelCredentials_missingCertChainFile_throws() throws IOException {
+    InstantiatingGrpcChannelProvider provider =
+        InstantiatingGrpcChannelProvider.newBuilder().build();
+    File trustBundle = new File("src/test/resources/root_cert.pem");
+    File privateKey = new File("src/test/resources/client_key.pem");
+    assertThat(provider.createMtlsToS2AChannelCredentials(trustBundle, privateKey, null)).isNull();
+  }
+
+  @Test
+  void createMtlsToS2AChannelCredentials_success() throws IOException {
+    InstantiatingGrpcChannelProvider provider =
+        InstantiatingGrpcChannelProvider.newBuilder().build();
+    File trustBundle = new File("src/test/resources/root_cert.pem");
+    File privateKey = new File("src/test/resources/client_key.pem");
+    File certChain = new File("src/test/resources/client_cert.pem");
+    assertEquals(
+        provider.createMtlsToS2AChannelCredentials(trustBundle, privateKey, certChain).getClass(),
+        TlsChannelCredentials.class);
+  }
+
+  @Test
+  void createS2ASecuredChannelCredentials_bothS2AAddressesNull_returnsNull() {
+    SecureSessionAgent s2aConfigProvider = Mockito.mock(SecureSessionAgent.class);
+    SecureSessionAgentConfig config = SecureSessionAgentConfig.createBuilder().build();
+    Mockito.when(s2aConfigProvider.getConfig()).thenReturn(config);
+    InstantiatingGrpcChannelProvider provider =
+        InstantiatingGrpcChannelProvider.newBuilder()
+            .setS2AConfigProvider(s2aConfigProvider)
+            .build();
+    assertThat(provider.createS2ASecuredChannelCredentials()).isNull();
+  }
+
+  @Test
+  void
+      createS2ASecuredChannelCredentials_mtlsS2AAddressNull_returnsPlaintextToS2AS2AChannelCredentials() {
+    SecureSessionAgent s2aConfigProvider = Mockito.mock(SecureSessionAgent.class);
+    SecureSessionAgentConfig config =
+        SecureSessionAgentConfig.createBuilder().setPlaintextAddress("localhost:8080").build();
+    Mockito.when(s2aConfigProvider.getConfig()).thenReturn(config);
+    FakeLogHandler logHandler = new FakeLogHandler();
+    InstantiatingGrpcChannelProvider.LOG.addHandler(logHandler);
+    InstantiatingGrpcChannelProvider provider =
+        InstantiatingGrpcChannelProvider.newBuilder()
+            .setS2AConfigProvider(s2aConfigProvider)
+            .build();
+    assertThat(provider.createS2ASecuredChannelCredentials()).isNotNull();
+    assertThat(logHandler.getAllMessages())
+        .contains(
+            "Cannot establish an mTLS connection to S2A because autoconfig endpoint did not return a mtls address to reach S2A.");
+    InstantiatingGrpcChannelProvider.LOG.removeHandler(logHandler);
+  }
+
+  @Test
+  void createS2ASecuredChannelCredentials_returnsPlaintextToS2AS2AChannelCredentials() {
+    SecureSessionAgent s2aConfigProvider = Mockito.mock(SecureSessionAgent.class);
+    SecureSessionAgentConfig config =
+        SecureSessionAgentConfig.createBuilder()
+            .setMtlsAddress("localhost:8080")
+            .setPlaintextAddress("localhost:8080")
+            .build();
+    Mockito.when(s2aConfigProvider.getConfig()).thenReturn(config);
+    FakeLogHandler logHandler = new FakeLogHandler();
+    InstantiatingGrpcChannelProvider.LOG.addHandler(logHandler);
+    InstantiatingGrpcChannelProvider provider =
+        InstantiatingGrpcChannelProvider.newBuilder()
+            .setS2AConfigProvider(s2aConfigProvider)
+            .build();
+    assertThat(provider.createS2ASecuredChannelCredentials()).isNotNull();
+    assertThat(logHandler.getAllMessages())
+        .contains(
+            "Cannot establish an mTLS connection to S2A because MTLS to MDS credentials do not exist on filesystem, falling back to plaintext connection to S2A");
+    InstantiatingGrpcChannelProvider.LOG.removeHandler(logHandler);
+  }
+
   private static class FakeLogHandler extends Handler {
 
     List<LogRecord> records = new ArrayList<>();

gax-java/gax-grpc/src/test/resources/README.md

@@ -0,0 +1,29 @@
+# Regenerate certificates and keys for testing mTLS-S2A
+Below are the commands which can be used to regenerate the certs used in tests. This is the same process
+used to generate test certs for S2A client in grpc-java: https://github.com/grpc/grpc-java/blob/master/s2a/src/test/resources/README.md
+
+Create root CA
+
+```
+openssl req -x509 -sha256 -days 7305 -newkey rsa:2048 -keyout root_key.pem -out
+root_cert.pem
+```
+
+Generate private key
+
+```
+openssl genrsa -out client_key.pem 2048
+```
+
+Generate CSR (set Common Name to localhost, leave all
+other fields blank)
+
+```
+openssl req -key client_key.pem -new -out client.csr -config config.cnf
+```
+
+Sign CSR for client
+
+```
+openssl x509 -req -CA root_cert.pem -CAkey root_key.pem -in client.csr -out client_cert.pem -days 7305
+```

gax-java/gax-grpc/src/test/resources/client_cert.pem

@@ -0,0 +1,20 @@
+-----BEGIN CERTIFICATE-----
+MIIDPTCCAiWgAwIBAgIUaarddwSWeE4jDC9kwxEr446ehqUwDQYJKoZIhvcNAQEL
+BQAwWTELMAkGA1UEBhMCQVUxEzARBgNVBAgMClNvbWUtU3RhdGUxITAfBgNVBAoM
+GEludGVybmV0IFdpZGdpdHMgUHR5IEx0ZDESMBAGA1UEAwwJbG9jYWxob3N0MB4X
+DTI0MTAwMTIxNTk1NFoXDTQ0MTAwMTIxNTk1NFowFDESMBAGA1UEAwwJbG9jYWxo
+b3N0MIIBIjANBgkqhkiG9w0BAQEFAAOCAQ8AMIIBCgKCAQEAxlNsldt7yAU4KRuS
+2D2/FjNIE1US5olBm4HteTr++41WaELZJqNLRPPp052jEQU3aKSYNGZvUUO6buu7
+eFpz2SBNUVMyvmzzocjVAyyf4NQvDazYHWOb+/YCeUppTRWriz4V5sn47qJTQ8cd
+CGrTFeLHxUjx4nh/OiqVXP/KnF3EqPEuqph0ky7+GirnJgPRe+C5ERuGkJye8dmP
+yWGA2lSS6MeDe7JZTAMi08bAn7BuNpeBkOzz1msGGI9PnUanUs7GOPWTDdcQAVY8
+KMvHCuGaNMGpb4rOR2mm8LlbAbpTPz8Pkw4QtMCLkgsrz2CzXpVwnLsU7nDXJAIO
+B155lQIDAQABo0IwQDAdBgNVHQ4EFgQUSZEyIHLzkIw7AwkBaUjYfIrGVR4wHwYD
+VR0jBBgwFoAUcq3dtxAVA410YWyM0B4e+4umbiwwDQYJKoZIhvcNAQELBQADggEB
+AAz0bZ4ayrZLhA45xn0yvdpdqiCtiWikCRtxgE7VXHg/ziZJVMpBpAhbIGO5tIyd
+lttnRXHwz5DUwKiba4/bCEFe229BshQEql5qaqcbGbFfSly11WeqqnwR1N7c8Gpv
+pD9sVrx22seN0rTUk87MY/S7mzCxHqAx35zm/LTW3pWcgCTMKFHy4Gt4mpTnXkNA
+WkhP2OhW5RLiu6Whi0BEdb2TGG1+ctamgijKXb+gJeef5ehlHXG8eU862KF5UlEA
+NeQKBm/PpQxOMe0NdpatjN8QRoczku0Itiodng+OZ1o+2iSNG988uFRb3CUSnjtE
+R/HL6ULAFzo59EpIYxruU/w=
+-----END CERTIFICATE-----

gax-java/gax-grpc/src/test/resources/client_key.pem

@@ -0,0 +1,28 @@
+-----BEGIN PRIVATE KEY-----
+MIIEvgIBADANBgkqhkiG9w0BAQEFAASCBKgwggSkAgEAAoIBAQDGU2yV23vIBTgp
+G5LYPb8WM0gTVRLmiUGbge15Ov77jVZoQtkmo0tE8+nTnaMRBTdopJg0Zm9RQ7pu
+67t4WnPZIE1RUzK+bPOhyNUDLJ/g1C8NrNgdY5v79gJ5SmlNFauLPhXmyfjuolND
+xx0IatMV4sfFSPHieH86KpVc/8qcXcSo8S6qmHSTLv4aKucmA9F74LkRG4aQnJ7x
+2Y/JYYDaVJLox4N7sllMAyLTxsCfsG42l4GQ7PPWawYYj0+dRqdSzsY49ZMN1xAB
+Vjwoy8cK4Zo0walvis5HaabwuVsBulM/Pw+TDhC0wIuSCyvPYLNelXCcuxTucNck
+Ag4HXnmVAgMBAAECggEAKuW9jXaBgiS63o1jyFkmvWcPNntG0M2sfrXuRzQfFgse
+vwOCk8xrSflWQNsOe+58ayp6746ekl3LdBWSIbiy6SqG/sm3pp/LXNmjVYHv/QH4
+QYV643R5t1ihdVnGiBFhXwdpVleme/tpdjYZzgnJKak5W69o/nrgzhSK5ShAy2xM
+j0XXbgdqG+4JxPb5BZmjHHfXAXUfgSORMdfArkbgFBRc9wL/6JVTXjeAMy5WX9qe
+5UQsSOYkwc9P2snifC/jdIhjHQOkkx59O0FgukJEFZPoagVG1duWQbnNDr7QVHCJ
+jV6dg9tIT4SXD3uPSPbgNGlRUseIakCzrhHARJuA2wKBgQD/h8zoh0KaqKyViCYw
+XKOFpm1pAFnp2GiDOblxNubNFAXEWnC+FlkvO/z1s0zVuYELUqfxcYMSXJFEVelK
+rfjZtoC5oxqWGqLo9iCj7pa8t+ipulYcLt2SWc7eZPD4T4lzeEf1Qz77aKcz34sa
+dv9lzQkDvhR/Mv1VeEGFHiq2VwKBgQDGsLcTGH5Yxs//LRSY8TigBkQEDrH5NvXu
+2jtAzZhy1Yhsoa5eiZkhnnzM6+n05ovfZLcy6s7dnwP1Y+C79vs+DKMBsodtDG5z
+YpsB0VrXYa6P6pCqkcz0Bz9xdo5sOhAK3AKnX6jd29XBDdeYsw/lxHLG24wProTD
+cCYFqtaj8wKBgQCaqKT68DL9zK14a8lBaDCIyexaqx3AjXzkP+Hfhi03XrEG4P5v
+7rLYBeTbCUSt7vMN2V9QoTWFvYUm6SCkVJvTmcRblz6WL1T+z0l+LwAJBP7LC77m
+m+77j2PH8yxt/iXhP6G97o+GNxdMLDbTM8bs5KZaH4fkXQY73uc5HMMZTQKBgEZS
+7blYhf+t/ph2wD+RwVUCYrh86wkmJs2veCFro3WhlnO8lhbn5Mc9bTaqmVgQ8ZjT
+8POYoDdYvPHxs+1TcYF4v4kuQziZmc5FLE/sZZauADb38tQsXrpQhmgGakpsEpmF
+XXsYJJDB6lo2KATn+8x7R5SSyHQUdPEnlI2U9ft5AoGBAJw0NJiM1EzRS8xq0DmO
+AvQaPjo01o2hH6wghws8gDQwrj0eHraHgVi7zo0VkaHJbO7ahKPudset3N7owJhA
+CUAPPRtv5wn0amAyNz77f1dz4Gys3AkcchflqhbEaQpzKYx4kX0adclur4WJ/DVm
+P7DI977SHCVB4FVMbXMEkBjN
+-----END PRIVATE KEY-----

gax-java/gax-grpc/src/test/resources/root_cert.pem

@@ -0,0 +1,22 @@
+-----BEGIN CERTIFICATE-----
+MIIDkzCCAnugAwIBAgIUWemeXZdfqcqkP8/Eyj74oTJtoNQwDQYJKoZIhvcNAQEL
+BQAwWTELMAkGA1UEBhMCQVUxEzARBgNVBAgMClNvbWUtU3RhdGUxITAfBgNVBAoM
+GEludGVybmV0IFdpZGdpdHMgUHR5IEx0ZDESMBAGA1UEAwwJbG9jYWxob3N0MB4X
+DTI0MTAwMTIxNTkxMVoXDTQ0MTAwMTIxNTkxMVowWTELMAkGA1UEBhMCQVUxEzAR
+BgNVBAgMClNvbWUtU3RhdGUxITAfBgNVBAoMGEludGVybmV0IFdpZGdpdHMgUHR5
+IEx0ZDESMBAGA1UEAwwJbG9jYWxob3N0MIIBIjANBgkqhkiG9w0BAQEFAAOCAQ8A
+MIIBCgKCAQEAt3A04hy5lljv86Nu0LLQZ2hA+fcImHjt1p1Mxgcta/5oxfVLcerE
+ZH+DAQLDtWzp9Up/vI57MM419GIL8Iszk7hnZRS/HWJ+2jewZJtz4i/g15dLr6+1
+uabMdPOWos60BwcLMxKEe6lJO1mV4z9d4NH4mAuMIHyM+ty0Klp9MfeDJtYEh0+z
+AxJUHCixDTsnKJro7My7A3ZT7bvaMfXxS7XN6qlRgBfiCmXo/GKTFfmfBW/EZGkG
+XOCxE2D79wYNhC41Q/ix0kwjEeOj2vgGFoiyblSdHdzvRXzsoQTEiZSM8lJDR2IT
+ZbpgbBlknMU6efNWlS8P5damB9ZWXg3x4wIDAQABo1MwUTAdBgNVHQ4EFgQUcq3d
+txAVA410YWyM0B4e+4umbiwwHwYDVR0jBBgwFoAUcq3dtxAVA410YWyM0B4e+4um
+biwwDwYDVR0TAQH/BAUwAwEB/zANBgkqhkiG9w0BAQsFAAOCAQEApZvaI9y7vjX/
+RRdvwf2Db9KlTE9nuVQ3AsrmG9Ml0p2X6U5aTetxdYBo2PuaaYHheF03JOH8zjpL
+UfFzvbi52DPbfFAaDw/6NIAenXlg492leNvUFNjGGRyJO9R5/aDfv40/fT3Em5G5
+DnR8SeGQ9tI1t6xBBT+d+/MilSiEKVu8IIF/p0SwvEyR4pKo6wFVZR0ZiIj2v/FZ
+P5Qk0Xhb+slpmaR3Wtx/mPl9Wb3kpPD4CAwhWDqFkKJql9/n9FvMjdwlCQKQGB26
+ZDXY3C0UTdktK5biNWRgAUVJEWBX6Q2amrxQHIn2d9RJ8uxCME/KBAntK+VxZE78
+w0JOvQ4Dpw==
+-----END CERTIFICATE-----

gax-java/gax-httpjson/clirr-ignored-differences.xml

@@ -106,15 +106,15 @@
     <className>com/google/api/gax/batching/Batcher</className>
     <method>*</method>
   </difference>
-  <!-- Ignore this as this was part of s2a-grpc ExperimentalApi revert -->
+  <!-- Ignore abstract method addition to an EndpointContext (InternalApi) -->
   <difference>
-    <differenceType>7002</differenceType>
-    <className>com/google/api/gax/rpc/FixedTransportChannelProvider</className>
-    <method>* withUseS2A(*)</method>
+    <differenceType>7013</differenceType>
+    <className>com/google/api/gax/rpc/EndpointContext</className>
+    <method>* useS2A()</method>
   </difference>
-  <!-- Ignore this as this was part of s2a-grpc ExperimentalApi revert -->
+  <!-- Ignore method addition to TransportChannelProvider interface (InternalExtensionOnly) -->
   <difference>
-    <differenceType>7002</differenceType>
+    <differenceType>7012</differenceType>
     <className>com/google/api/gax/rpc/TransportChannelProvider</className>
     <method>* withUseS2A(*)</method>
   </difference>

gax-java/gax/clirr-ignored-differences.xml

@@ -222,6 +222,7 @@ public static ClientContext create(StubSettings settings) throws IOException {
     if (transportChannelProvider.needsEndpoint()) {
       transportChannelProvider = transportChannelProvider.withEndpoint(endpoint);
     }
+    transportChannelProvider = transportChannelProvider.withUseS2A(endpointContext.useS2A());
     TransportChannel transportChannel = transportChannelProvider.getTransportChannel();
 
     ApiCallContext defaultCallContext =

gax-java/gax/src/main/java/com/google/api/gax/rpc/ClientContext.java

@@ -30,6 +30,7 @@
 package com.google.api.gax.rpc;
 
 import com.google.api.core.InternalApi;
+import com.google.api.gax.rpc.internal.EnvironmentProvider;
 import com.google.api.gax.rpc.mtls.MtlsProvider;
 import com.google.auth.Credentials;
 import com.google.auth.oauth2.ComputeEngineCredentials;
@@ -65,6 +66,9 @@ public abstract class EndpointContext {
       "The configured universe domain (%s) does not match the universe domain found in the credentials (%s). If you haven't configured the universe domain explicitly, `googleapis.com` is the default.";
   public static final String UNABLE_TO_RETRIEVE_CREDENTIALS_ERROR_MESSAGE =
       "Unable to retrieve the Universe Domain from the Credentials.";
+  // This environment variable is a temporary measure. It will be removed when the feature is
+  // non-experimental.
+  static final String S2A_ENV_ENABLE_USE_S2A = "EXPERIMENTAL_GOOGLE_API_USE_S2A_JAVA";
 
   public static EndpointContext getDefaultInstance() {
     return INSTANCE;
@@ -100,6 +104,11 @@ public static EndpointContext getDefaultInstance() {
   @Nullable
   public abstract String transportChannelProviderEndpoint();
 
+  abstract boolean useS2A();
+
+  @Nullable
+  abstract EnvironmentProvider envProvider();
+
   @Nullable
   public abstract String mtlsEndpoint();
 
@@ -119,7 +128,8 @@ public static EndpointContext getDefaultInstance() {
   public static Builder newBuilder() {
     return new AutoValue_EndpointContext.Builder()
         .setSwitchToMtlsEndpointAllowed(false)
-        .setUsingGDCH(false);
+        .setUsingGDCH(false)
+        .setEnvProvider(System::getenv);
   }
 
   /** Configure the existing EndpointContext to be using GDC-H */
@@ -208,6 +218,10 @@ public abstract static class Builder {
 
     public abstract Builder setResolvedUniverseDomain(String resolvedUniverseDomain);
 
+    abstract Builder setUseS2A(boolean useS2A);
+
+    abstract Builder setEnvProvider(EnvironmentProvider envProvider);
+
     abstract String serviceName();
 
     abstract String universeDomain();
@@ -216,6 +230,10 @@ public abstract static class Builder {
 
     abstract String transportChannelProviderEndpoint();
 
+    abstract boolean useS2A();
+
+    abstract EnvironmentProvider envProvider();
+
     abstract String mtlsEndpoint();
 
     abstract boolean switchToMtlsEndpointAllowed();
@@ -285,9 +303,46 @@ private String determineEndpoint() throws IOException {
             "mTLS is not supported in any universe other than googleapis.com");
       }
 
+      // Check if Experimental S2A feature enabled. When feature is non-experimental, remove this
+      // check from this function, and plumb MTLS endpoint to channel creation logic separately.
+      // Note that mTLS via S2A is an independent feature from mTLS via DCA (for which endpoint
+      // determined by {@code mtlsEndpointResolver} above).
+      if (shouldUseS2A()) {
+        return mtlsEndpoint();
+      }
       return endpoint;
     }
 
+    /** Determine if S2A can be used */
+    @VisibleForTesting
+    boolean shouldUseS2A() {
+      // If mTLS endpoint is not available, skip S2A
+      if (Strings.isNullOrEmpty(mtlsEndpoint())) {
+        return false;
+      }
+
+      // If EXPERIMENTAL_GOOGLE_API_USE_S2A_JAVA is not set to true, skip S2A.
+      String s2AEnv = envProvider().getenv(S2A_ENV_ENABLE_USE_S2A);
+      boolean s2AEnabled = Boolean.parseBoolean(s2AEnv);
+      if (!s2AEnabled) {
+        return false;
+      }
+
+      // Skip S2A when using GDC-H
+      if (usingGDCH()) {
+        return false;
+      }
+
+      // If a custom endpoint is being used, skip S2A.
+      if (!Strings.isNullOrEmpty(clientSettingsEndpoint())
+          || !Strings.isNullOrEmpty(transportChannelProviderEndpoint())) {
+        return false;
+      }
+
+      // mTLS via S2A is not supported in any universe other than googleapis.com.
+      return mtlsEndpoint().contains(Credentials.GOOGLE_DEFAULT_UNIVERSE);
+    }
+
     // Default to port 443 for HTTPS. Using HTTP requires explicitly setting the endpoint
     private String buildEndpointTemplate(String serviceName, String resolvedUniverseDomain) {
       return serviceName + "." + resolvedUniverseDomain + ":443";
@@ -321,6 +376,7 @@ public EndpointContext build() throws IOException {
       // The Universe Domain is used to resolve the Endpoint. It should be resolved first
       setResolvedUniverseDomain(determineUniverseDomain());
       setResolvedEndpoint(determineEndpoint());
+      setUseS2A(shouldUseS2A());
       return autoBuild();
     }
   }

gax-java/gax/src/main/java/com/google/api/gax/rpc/EndpointContext.java

@@ -97,6 +97,13 @@ public interface TransportChannelProvider {
    */
   TransportChannelProvider withEndpoint(String endpoint);
 
+  /** Sets whether to use S2A when constructing a new {@link TransportChannel}. */
+  @BetaApi(
+      "The S2A feature is not stable yet and may change in the future. https://github.com/grpc/grpc-java/issues/11533.")
+  default TransportChannelProvider withUseS2A(boolean useS2A) {
+    return this;
+  }
+
   /**
    * Reports whether this provider allows pool size customization.
    *

gax-java/gax/src/main/java/com/google/api/gax/rpc/TransportChannelProvider.java

@@ -33,6 +33,7 @@
 import static org.junit.jupiter.api.Assertions.assertThrows;
 
 import com.google.api.gax.core.NoCredentialsProvider;
+import com.google.api.gax.rpc.internal.EnvironmentProvider;
 import com.google.api.gax.rpc.mtls.MtlsProvider;
 import com.google.api.gax.rpc.testing.FakeMtlsProvider;
 import com.google.auth.Credentials;
@@ -372,6 +373,21 @@ void endpointContextBuild_multipleUniverseDomainConfigurations_clientSettingsHas
         .isEqualTo(clientSettingsUniverseDomain);
   }
 
+  @Test
+  void endpointContextBuild_shouldUseS2A_mtlsEndpoint() throws IOException {
+    EnvironmentProvider envProvider = Mockito.mock(EnvironmentProvider.class);
+    Mockito.when(envProvider.getenv(EndpointContext.S2A_ENV_ENABLE_USE_S2A)).thenReturn("true");
+    defaultEndpointContextBuilder =
+        defaultEndpointContextBuilder
+            .setEnvProvider(envProvider)
+            .setClientSettingsEndpoint("")
+            .setTransportChannelProviderEndpoint("")
+            .setUsingGDCH(false);
+    EndpointContext endpointContext = defaultEndpointContextBuilder.build();
+    Truth.assertThat(defaultEndpointContextBuilder.shouldUseS2A()).isTrue();
+    Truth.assertThat(endpointContext.resolvedEndpoint()).isEqualTo(DEFAULT_MTLS_ENDPOINT);
+  }
+
   @Test
   void hasValidUniverseDomain_gdchFlow_anyCredentials() throws IOException {
     Credentials noCredentials = NoCredentialsProvider.create().getCredentials();
@@ -454,4 +470,111 @@ void hasValidUniverseDomain_computeEngineCredentials_noValidationOnUniverseDomai
             .build();
     assertDoesNotThrow(() -> endpointContext.validateUniverseDomain(credentials, statusCode));
   }
+
+  @Test
+  void shouldUseS2A_envVarNotSet_returnsFalse() throws IOException {
+    EnvironmentProvider envProvider = Mockito.mock(EnvironmentProvider.class);
+    Mockito.when(envProvider.getenv(EndpointContext.S2A_ENV_ENABLE_USE_S2A)).thenReturn("false");
+    defaultEndpointContextBuilder =
+        defaultEndpointContextBuilder
+            .setEnvProvider(envProvider)
+            .setClientSettingsEndpoint("")
+            .setTransportChannelProviderEndpoint("")
+            .setUsingGDCH(false);
+    Truth.assertThat(defaultEndpointContextBuilder.shouldUseS2A()).isFalse();
+  }
+
+  @Test
+  void shouldUseS2A_UsingGDCH_returnsFalse() throws IOException {
+    EnvironmentProvider envProvider = Mockito.mock(EnvironmentProvider.class);
+    Mockito.when(envProvider.getenv(EndpointContext.S2A_ENV_ENABLE_USE_S2A)).thenReturn("true");
+    defaultEndpointContextBuilder =
+        defaultEndpointContextBuilder
+            .setEnvProvider(envProvider)
+            .setClientSettingsEndpoint("")
+            .setTransportChannelProviderEndpoint("")
+            .setUsingGDCH(true);
+    Truth.assertThat(defaultEndpointContextBuilder.shouldUseS2A()).isFalse();
+  }
+
+  @Test
+  void shouldUseS2A_customEndpointSetViaClientSettings_returnsFalse() throws IOException {
+    EnvironmentProvider envProvider = Mockito.mock(EnvironmentProvider.class);
+    Mockito.when(envProvider.getenv(EndpointContext.S2A_ENV_ENABLE_USE_S2A)).thenReturn("true");
+    defaultEndpointContextBuilder =
+        defaultEndpointContextBuilder
+            .setEnvProvider(envProvider)
+            .setClientSettingsEndpoint("test.endpoint.com:443")
+            .setTransportChannelProviderEndpoint("")
+            .setUsingGDCH(false);
+    Truth.assertThat(defaultEndpointContextBuilder.shouldUseS2A()).isFalse();
+  }
+
+  @Test
+  void shouldUseS2A_customEndpointSetViaTransportChannelProvider_returnsFalse() throws IOException {
+    EnvironmentProvider envProvider = Mockito.mock(EnvironmentProvider.class);
+    Mockito.when(envProvider.getenv(EndpointContext.S2A_ENV_ENABLE_USE_S2A)).thenReturn("true");
+    defaultEndpointContextBuilder =
+        defaultEndpointContextBuilder
+            .setEnvProvider(envProvider)
+            .setClientSettingsEndpoint("")
+            .setTransportChannelProviderEndpoint("test.endpoint.com:443")
+            .setUsingGDCH(false);
+    Truth.assertThat(defaultEndpointContextBuilder.shouldUseS2A()).isFalse();
+  }
+
+  @Test
+  void shouldUseS2A_mtlsEndpointNull_returnsFalse() throws IOException {
+    EnvironmentProvider envProvider = Mockito.mock(EnvironmentProvider.class);
+    Mockito.when(envProvider.getenv(EndpointContext.S2A_ENV_ENABLE_USE_S2A)).thenReturn("true");
+    defaultEndpointContextBuilder =
+        defaultEndpointContextBuilder
+            .setEnvProvider(envProvider)
+            .setClientSettingsEndpoint("")
+            .setTransportChannelProviderEndpoint("")
+            .setUsingGDCH(false)
+            .setMtlsEndpoint(null);
+    Truth.assertThat(defaultEndpointContextBuilder.shouldUseS2A()).isFalse();
+  }
+
+  @Test
+  void shouldUseS2A_mtlsEndpointEmpty_returnsFalse() throws IOException {
+    EnvironmentProvider envProvider = Mockito.mock(EnvironmentProvider.class);
+    Mockito.when(envProvider.getenv(EndpointContext.S2A_ENV_ENABLE_USE_S2A)).thenReturn("true");
+    defaultEndpointContextBuilder =
+        defaultEndpointContextBuilder
+            .setEnvProvider(envProvider)
+            .setClientSettingsEndpoint("")
+            .setTransportChannelProviderEndpoint("")
+            .setMtlsEndpoint("")
+            .setUsingGDCH(false);
+    Truth.assertThat(defaultEndpointContextBuilder.shouldUseS2A()).isFalse();
+  }
+
+  @Test
+  void shouldUseS2A_mtlsEndpointNotGoogleDefaultUniverse_returnsFalse() throws IOException {
+    EnvironmentProvider envProvider = Mockito.mock(EnvironmentProvider.class);
+    Mockito.when(envProvider.getenv(EndpointContext.S2A_ENV_ENABLE_USE_S2A)).thenReturn("true");
+    defaultEndpointContextBuilder =
+        defaultEndpointContextBuilder
+            .setEnvProvider(envProvider)
+            .setClientSettingsEndpoint("")
+            .setTransportChannelProviderEndpoint("")
+            .setMtlsEndpoint("test.mtls.abcd.com:443")
+            .setUsingGDCH(false);
+    Truth.assertThat(defaultEndpointContextBuilder.shouldUseS2A()).isFalse();
+  }
+
+  @Test
+  void shouldUseS2A_success() throws IOException {
+    EnvironmentProvider envProvider = Mockito.mock(EnvironmentProvider.class);
+    Mockito.when(envProvider.getenv(EndpointContext.S2A_ENV_ENABLE_USE_S2A)).thenReturn("true");
+    defaultEndpointContextBuilder =
+        defaultEndpointContextBuilder
+            .setEnvProvider(envProvider)
+            .setClientSettingsEndpoint("")
+            .setTransportChannelProviderEndpoint("")
+            .setUsingGDCH(false);
+    Truth.assertThat(defaultEndpointContextBuilder.shouldUseS2A()).isTrue();
+  }
 }