@@ -34,12 +34,17 @@
 import com.google.auth.oauth2.EnvironmentProvider;
 import com.google.auth.oauth2.PropertyProvider;
 import com.google.common.base.Strings;
+import com.google.common.io.BaseEncoding;
 import java.io.File;
 import java.io.FileInputStream;
 import java.io.IOException;
 import java.io.InputStream;
+import java.nio.file.Files;
+import java.nio.file.Paths;
+import java.security.MessageDigest;
 import java.util.Locale;
 import org.jspecify.annotations.NullMarked;
+import org.jspecify.annotations.Nullable;
 
 /**
  * Utility class for mTLS related operations.
@@ -57,6 +62,168 @@ private MtlsUtils() {
     // Prevent instantiation for Utility class
   }
 
+  /**
+   * Returns if mutual TLS client certificate should be used. Returns true if valid workload
+   * certificates are configured or if GOOGLE_API_USE_CLIENT_CERTIFICATE is explicitly set to true
+   * (e.g. for Enterprise Certificate Proxy or custom MtlsProviders), unless explicitly disabled via
+   * GOOGLE_API_USE_CLIENT_CERTIFICATE=false.
+   */
+  public static boolean useMtlsClientCertificate(
+      EnvironmentProvider envProvider, PropertyProvider propProvider) {
+    String useClientCertificate = envProvider.getEnv("GOOGLE_API_USE_CLIENT_CERTIFICATE");
+    if ("false".equalsIgnoreCase(useClientCertificate)) {
+      return false;
+    }
+    if (getWorkloadCertPath(envProvider, propProvider) != null) {
+      return true;
+    }
+    return "true".equalsIgnoreCase(useClientCertificate);
+  }
+
+  /**
+   * Resolves and returns the path to the mutual TLS client certificate, or null if none should be
+   * used.
+   *
+   * <p>Possible outcomes:
+   *
+   * <ol>
+   *   <li><b>Non-null {@link String} (Valid happy path):</b> A valid workload certificate
+   *       configuration was found and both the certificate and private key files exist and are
+   *       readable.
+   *   <li><b>{@link IllegalStateException} (Invalid state - fail closed):</b> An explicit {@code
+   *       GOOGLE_API_CERTIFICATE_CONFIG} path or an existing default well-known certificate
+   *       configuration file is missing, unreadable, malformed, or references missing/unreadable
+   *       certificate or private key files. This is treated as an unrecoverable misconfiguration.
+   *   <li><b>{@code null} (Safe fallback / fail open):</b> Client certificates are explicitly
+   *       disabled via {@code GOOGLE_API_USE_CLIENT_CERTIFICATE=false}, no explicit configuration
+   *       is set and the default well-known configuration file does not exist on disk, or the
+   *       configuration specifies an non-workload source (e.g., ECP/PKCS11 without a {@code
+   *       workload} section). Callers can proceed without workload certificate file polling.
+   * </ol>
+   */
+  public static @Nullable String getWorkloadCertPath(
+      EnvironmentProvider envProvider, PropertyProvider propProvider) {
+    String useClientCertificate = envProvider.getEnv("GOOGLE_API_USE_CLIENT_CERTIFICATE");
+    if ("false".equalsIgnoreCase(useClientCertificate)) {
+      return null;
+    }
+
+    String explicitConfigPath = envProvider.getEnv(CERTIFICATE_CONFIGURATION_ENV_VARIABLE);
+
+    // 1. Explicit Configuration Path (Fail Closed)
+    if (!Strings.isNullOrEmpty(explicitConfigPath)) {
+      File configFile = new File(explicitConfigPath);
+      if (!configFile.exists()) {
+        throw new IllegalStateException(
+            "Certificate configuration file specified via GOOGLE_API_CERTIFICATE_CONFIG at '"
+                + explicitConfigPath
+                + "' does not exist.");
+      }
+      if (!configFile.isFile() || !configFile.canRead()) {
+        throw new IllegalStateException(
+            "Failed to read certificate configuration file specified via"
+                + " GOOGLE_API_CERTIFICATE_CONFIG at '"
+                + explicitConfigPath
+                + "'.");
+      }
+      WorkloadCertificateConfiguration config;
+      try {
+        config = getWorkloadCertificateConfiguration(envProvider, propProvider, explicitConfigPath);
+      } catch (CertificateSourceUnavailableException e) {
+        // ECP / PKCS11 configuration without workload section; safe fallback
+        return null;
+      } catch (Exception e) {
+        throw new IllegalStateException(
+            "Certificate configuration file specified via GOOGLE_API_CERTIFICATE_CONFIG at '"
+                + explicitConfigPath
+                + "' is malformed: "
+                + e.getMessage(),
+            e);
+      }
+      checkCertAndKeyFilesReadable(config, explicitConfigPath, false);
+      return config.getCertPath();
+    }
+
+    // 2. Implicit / Default gcloud Configuration Path
+    File defaultConfigFile = null;
+    try {
+      defaultConfigFile = getWellKnownCertificateConfigFile(envProvider, propProvider);
+    } catch (IOException e) {
+      // APPDATA missing on Windows, etc. Safe fallback.
+    }
+    if (defaultConfigFile != null && defaultConfigFile.exists()) {
+      if (!defaultConfigFile.isFile() || !defaultConfigFile.canRead()) {
+        throw new IllegalStateException(
+            "Default certificate configuration file at '"
+                + defaultConfigFile.getAbsolutePath()
+                + "' exists but could not be read.");
+      }
+      WorkloadCertificateConfiguration config = null;
+      try {
+        config = getWorkloadCertificateConfiguration(envProvider, propProvider, null);
+      } catch (CertificateSourceUnavailableException e) {
+        // ECP-only configuration without workload section; safe fallback
+      } catch (Exception e) {
+        throw new IllegalStateException(
+            "Default certificate configuration file at '"
+                + defaultConfigFile.getAbsolutePath()
+                + "' is malformed: "
+                + e.getMessage(),
+            e);
+      }
+      if (config != null) {
+        checkCertAndKeyFilesReadable(config, defaultConfigFile.getAbsolutePath(), true);
+        return config.getCertPath();
+      }
+    }
+
+    return null;
+  }
+
+  private static void checkCertAndKeyFilesReadable(
+      WorkloadCertificateConfiguration config, String configPath, boolean isDefaultConfig) {
+    File certFile = new File(config.getCertPath());
+    File keyFile = new File(config.getPrivateKeyPath());
+    if (!certFile.isFile() || !certFile.canRead() || !keyFile.isFile() || !keyFile.canRead()) {
+      String sourcePrefix =
+          isDefaultConfig
+              ? "referenced by default configuration '"
+              : "referenced by configuration '";
+      throw new IllegalStateException(
+          "Failed to read certificate/key file at '"
+              + config.getCertPath()
+              + "' or '"
+              + config.getPrivateKeyPath()
+              + "' "
+              + sourcePrefix
+              + configPath
+              + "'.");
+    }
+  }
+
+  /**
+   * Computes the lower-case SHA-256 hex fingerprint of the certificate file at {@code certPath}.
+   *
+   * <p>Unlike {@link #getWorkloadCertPath}, which validates configuration at channel initialization
+   * and fails closed on errors, this method is called dynamically at runtime during active RPCs to
+   * detect certificate rotations on disk. External certificate rotators may temporarily delete,
+   * truncate, or rewrite the certificate file mid-RPC. Returning {@code null} on read/digest
+   * exceptions (which callers normalize to {@code ""}) allows runtime refresh checks to ignore
+   * transient mid-write states and keep the active healthy channel without failing in-flight RPCs.
+   */
+  public static @Nullable String getCertificateFingerprint(@Nullable String certPath) {
+    if (certPath == null) {
+      return null;
+    }
+    try {
+      byte[] certBytes = Files.readAllBytes(Paths.get(certPath));
+      byte[] digest = MessageDigest.getInstance("SHA-256").digest(certBytes);
+      return BaseEncoding.base16().lowerCase().encode(digest);
+    } catch (Exception e) {
+      return null;
+    }
+  }
+
   /**
    * Returns the path to the client certificate file specified by the loaded workload certificate
    * configuration.
@@ -65,14 +232,17 @@ private MtlsUtils() {
    * @throws IOException if the certificate configuration cannot be found or loaded.
    */
   public static String getCertificatePath(
-      EnvironmentProvider envProvider, PropertyProvider propProvider, String certConfigPathOverride)
+      EnvironmentProvider envProvider,
+      PropertyProvider propProvider,
+      @Nullable String certConfigPathOverride)
       throws IOException {
     String certPath =
         getWorkloadCertificateConfiguration(envProvider, propProvider, certConfigPathOverride)
             .getCertPath();
     if (Strings.isNullOrEmpty(certPath)) {
       throw new CertificateSourceUnavailableException(
-          "Certificate configuration loaded successfully, but does not contain a 'certificate_file' path.");
+          "Certificate configuration loaded successfully, but does not contain a"
+              + " 'cert_configs.workload.cert_path' path.");
     }
     return certPath;
   }
@@ -92,7 +262,9 @@ public static String getCertificatePath(
    * @throws IOException if the configuration file cannot be found, read, or parsed
    */
   static WorkloadCertificateConfiguration getWorkloadCertificateConfiguration(
-      EnvironmentProvider envProvider, PropertyProvider propProvider, String certConfigPathOverride)
+      EnvironmentProvider envProvider,
+      PropertyProvider propProvider,
+      @Nullable String certConfigPathOverride)
       throws IOException {
     File certConfig;
     if (certConfigPathOverride != null) {
@@ -106,7 +278,7 @@ static WorkloadCertificateConfiguration getWorkloadCertificateConfiguration(
       }
     }
 
-    if (!certConfig.isFile()) {
+    if (!certConfig.isFile() || !certConfig.canRead()) {
       throw new CertificateSourceUnavailableException(
           "Certificate configuration file does not exist or is not a file: "
               + certConfig.getAbsolutePath());