fix(auth): align mTLS discovery and enforce fail-fast transport configuration.

packages/google-auth/google/auth/_agent_identity_utils.py

@@ -201,6 +201,13 @@ def get_and_parse_agent_identity_certificate():
     if is_opted_out:
         return None
 
+    # Respect explicit opt-out of mTLS / client certs
+    from google.auth.transport import _mtls_helper
+
+    env_override = _mtls_helper._check_use_client_cert_env()
+    if env_override is False:
+        return None
+
     cert_path = get_agent_identity_certificate_path()
     if not cert_path:
         return None
@@ -312,7 +319,17 @@ def should_request_bound_token(cert):
         ).lower()
         == "true"
     )
-    return is_agent_cert and is_opted_in
+    if not (is_agent_cert and is_opted_in):
+        return False
+
+    # Respect explicit opt-out of mTLS / client certs
+    from google.auth.transport import _mtls_helper
+
+    env_override = _mtls_helper._check_use_client_cert_env()
+    if env_override is False:
+        return False
+
+    return True
 
 
 def get_cached_cert_fingerprint(cached_cert):

packages/google-auth/google/auth/aio/transport/sessions.py

@@ -17,6 +17,7 @@
 import functools
 import time
 from typing import Mapping, Optional, TYPE_CHECKING, Union
+import warnings
 
 from google.auth import _exponential_backoff, exceptions
 from google.auth.aio import transport
@@ -36,6 +37,7 @@
     except (ImportError, AttributeError):
         ClientTimeout = None
 
+
 # Tracks the internal aiohttp installation and usage
 try:
     from google.auth.aio.transport.aiohttp import Request as AiohttpRequest
@@ -150,11 +152,11 @@ def __init__(
     async def configure_mtls_channel(self, client_cert_callback=None):
         """Configure the client certificate and key for SSL connection.
 
-        The function does nothing unless `GOOGLE_API_USE_CLIENT_CERTIFICATE` is
-        explicitly set to `true`. In this case if client certificate and key are
-        successfully obtained (from the given client_cert_callback or from application
-        default SSL credentials), the underlying transport will be reconfigured
-        to use mTLS.
+        This method configures mTLS if client certificates are explicitly enabled
+        (via GOOGLE_API_USE_CLIENT_CERTIFICATE=true) or auto-enabled (when the env
+        variable is unset and workload certificates are discovered). In these cases,
+        the underlying transport will be reconfigured to use mTLS.
+
         Note: This function does nothing if the `aiohttp` library is not
         installed.
         Important: Calling this method will close any ongoing API requests associated
@@ -207,12 +209,23 @@ async def _do_configure():
                             self._auth_request = AiohttpRequest(session=new_session)
 
                             await old_auth_request.close()
+                        else:
+                            self._is_mtls = False
+                            warnings.warn(
+                                "Attempted to establish mTLS, but a custom async transport was provided. "
+                                "google-auth cannot automatically configure custom transports for mTLS. "
+                                "Falling back to standard TLS. If your custom transport is not manually "
+                                "configured for mTLS, you may encounter 401 Unauthorized errors when "
+                                "using Certificate-Bound Tokens.",
+                                UserWarning,
+                            )
 
                 except (
                     exceptions.ClientCertError,
                     ImportError,
                     OSError,
                 ) as caught_exc:
+                    self._is_mtls = False
                     new_exc = exceptions.MutualTLSChannelError(caught_exc)
                     raise new_exc from caught_exc
 

packages/google-auth/google/auth/transport/_mtls_helper.py

@@ -46,6 +46,12 @@
 _LOGGER = logging.getLogger(__name__)
 
 
+# A flag to track if we have already logged a warning about mTLS auto-enablement failures.
+# This prevents log spam when client libraries create transports or session instances
+# frequently within a single process.
+_has_logged_mtls_warning = False
+
+
 _PASSPHRASE_REGEX = re.compile(
     b"-----BEGIN PASSPHRASE-----(.+)-----END PASSPHRASE-----", re.DOTALL
 )
@@ -200,12 +206,13 @@ def _get_workload_cert_and_key_paths(config_path, include_context_aware=True):
         return None, None
     workload = cert_configs["workload"]
 
-    if "cert_path" not in workload:
-        return None, None
+    if "cert_path" not in workload or "key_path" not in workload:
+        raise exceptions.ClientCertError(
+            'Workload certificate configuration is missing "cert_path" or "key_path" in {}'.format(
+                absolute_path
+            )
+        )
     cert_path = workload["cert_path"]
-
-    if "key_path" not in workload:
-        return None, None
     key_path = workload["key_path"]
 
     # == BEGIN Temporary Cloud Run PATCH ==
@@ -448,53 +455,70 @@ def client_cert_callback():
     return crypto.dump_privatekey(crypto.FILETYPE_PEM, pkey)
 
 
+def _check_use_client_cert_env():
+    use_client_cert = getenv(
+        environment_vars.GOOGLE_API_USE_CLIENT_CERTIFICATE
+    ) or getenv(environment_vars.CLOUDSDK_CONTEXT_AWARE_USE_CLIENT_CERTIFICATE)
+
+    if use_client_cert:
+        return use_client_cert.lower() == "true"
+    return None
+
+
 def check_use_client_cert():
     """Returns boolean for whether the client certificate should be used for mTLS.
 
     If GOOGLE_API_USE_CLIENT_CERTIFICATE is set to true or false, a corresponding
     bool value will be returned. If the value is set to an unexpected string, it
     will default to False.
     If GOOGLE_API_USE_CLIENT_CERTIFICATE is unset, the value will be inferred
-    by reading a file pointed at by GOOGLE_API_CERTIFICATE_CONFIG, and verifying
-    it contains a "workload" section. If so, the function will return True,
-    otherwise False.
+    as True (auto-enabled) if a workload config file exists (pointed at by
+    GOOGLE_API_CERTIFICATE_CONFIG) containing a "workload" section.
+    Otherwise, it returns False.
 
     Returns:
         bool: Whether the client certificate should be used for mTLS connection.
     """
-    use_client_cert = getenv(environment_vars.GOOGLE_API_USE_CLIENT_CERTIFICATE)
-    if use_client_cert is None or use_client_cert == "":
-        use_client_cert = getenv(
-            environment_vars.CLOUDSDK_CONTEXT_AWARE_USE_CLIENT_CERTIFICATE
-        )
+    global _has_logged_mtls_warning
+    env_override = _check_use_client_cert_env()
+    if env_override is not None:
+        return env_override
 
-    # Check if the value of GOOGLE_API_USE_CLIENT_CERTIFICATE is set.
-    if use_client_cert:
-        return use_client_cert.lower() == "true"
-    else:
-        # Check if the value of GOOGLE_API_CERTIFICATE_CONFIG is set.
-        cert_path = getenv(environment_vars.GOOGLE_API_CERTIFICATE_CONFIG)
-        if cert_path is None:
-            cert_path = getenv(
-                environment_vars.CLOUDSDK_CONTEXT_AWARE_CERTIFICATE_CONFIG_FILE_PATH
-            )
+    # Auto-enablement checks (when GOOGLE_API_USE_CLIENT_CERTIFICATE is not set)
+
+    # Check if the value of GOOGLE_API_CERTIFICATE_CONFIG is set.
+    cert_path = getenv(environment_vars.GOOGLE_API_CERTIFICATE_CONFIG) or getenv(
+        environment_vars.CLOUDSDK_CONTEXT_AWARE_CERTIFICATE_CONFIG_FILE_PATH
+    )
 
-        if cert_path:
-            try:
-                with open(cert_path, "r") as f:
-                    content = json.load(f)
-                    # verify json has workload key
-                    content["cert_configs"]["workload"]
-                    return True
-            except (
-                FileNotFoundError,
-                OSError,
-                KeyError,
-                TypeError,
-                json.JSONDecodeError,
-            ) as e:
-                _LOGGER.debug("error decoding certificate: %s", e)
-        return False
+    if cert_path:
+        try:
+            with open(cert_path, "r") as f:
+                content = json.load(f)
+        except (FileNotFoundError, OSError, json.JSONDecodeError) as e:
+            if not _has_logged_mtls_warning:
+                _LOGGER.warning(
+                    "mTLS auto-enablement failed: Could not read/parse certificate file at %s. Error: %s",
+                    cert_path,
+                    e,
+                )
+                _has_logged_mtls_warning = True
+            return False
+
+        # Structural validation
+        if isinstance(content, dict):
+            cert_configs = content.get("cert_configs")
+            if isinstance(cert_configs, dict) and "workload" in cert_configs:
+                return True
+
+        # If we got here, the file exists but the expected structure is missing
+        if not _has_logged_mtls_warning:
+            _LOGGER.warning(
+                "mTLS auto-enablement failed: Certificate configuration file at %s is missing the required ['cert_configs']['workload'] section.",
+                cert_path,
+            )
+            _has_logged_mtls_warning = True
+    return False
 
 
 def check_parameters_for_unauthorized_response(cached_cert):

packages/google-auth/google/auth/transport/requests.py

@@ -428,11 +428,11 @@ def __init__(
     def configure_mtls_channel(self, client_cert_callback=None):
         """Configure the client certificate and key for SSL connection.
 
-        The function does nothing unless `GOOGLE_API_USE_CLIENT_CERTIFICATE` is
-        explicitly set to `true`. In this case if client certificate and key are
-        successfully obtained (from the given client_cert_callback or from application
-        default SSL credentials), a :class:`_MutualTlsAdapter` instance will be mounted
-        to "https://" prefix.
+        This method configures mTLS if client certificates are explicitly enabled
+        (via GOOGLE_API_USE_CLIENT_CERTIFICATE=true) or auto-enabled (when the env
+        variable is unset and workload certificates are discovered). In these cases,
+        if the client certificate and key are successfully obtained, a
+        :class:`_MutualTlsAdapter` instance will be mounted to the "https://" prefix.
 
         Args:
             client_cert_callback (Optional[Callable[[], (bytes, bytes)]]):
@@ -452,6 +452,7 @@ def configure_mtls_channel(self, client_cert_callback=None):
         try:
             import OpenSSL
         except ImportError as caught_exc:
+            self._is_mtls = False
             new_exc = exceptions.MutualTLSChannelError(caught_exc)
             raise new_exc from caught_exc
 
@@ -471,8 +472,10 @@ def configure_mtls_channel(self, client_cert_callback=None):
         except (
             exceptions.ClientCertError,
             ImportError,
+            OSError,
             OpenSSL.crypto.Error,
         ) as caught_exc:
+            self._is_mtls = False
             new_exc = exceptions.MutualTLSChannelError(caught_exc)
             raise new_exc from caught_exc
 

packages/google-auth/google/auth/transport/urllib3.py

@@ -313,13 +313,12 @@ def __init__(
 
     def configure_mtls_channel(self, client_cert_callback=None):
         """Configures mutual TLS channel using the given client_cert_callback or
-        application default SSL credentials. The behavior is controlled by
-        `GOOGLE_API_USE_CLIENT_CERTIFICATE` environment variable.
-        (1) If the environment variable value is `true`, the function returns True
-        if the channel is mutual TLS and False otherwise. The `http` provided
-        in the constructor will be overwritten.
-        (2) If the environment variable is not set or `false`, the function does
-        nothing and it always return False.
+        application default SSL credentials.
+
+        The channel is configured if GOOGLE_API_USE_CLIENT_CERTIFICATE is "true",
+        or if it is unset and workload certificates are detected in the environment.
+        If client_cert_callback is None, default SSL credentials (workload or SecureConnect)
+        are loaded.
 
         Args:
             client_cert_callback (Optional[Callable[[], (bytes, bytes)]]):
@@ -344,6 +343,7 @@ def configure_mtls_channel(self, client_cert_callback=None):
         try:
             import OpenSSL
         except ImportError as caught_exc:
+            self._is_mtls = False
             new_exc = exceptions.MutualTLSChannelError(caught_exc)
             raise new_exc from caught_exc
 
@@ -357,11 +357,14 @@ def configure_mtls_channel(self, client_cert_callback=None):
                 self._cached_cert = cert
             else:
                 self.http = _make_default_http()
+                self._is_mtls = False
         except (
             exceptions.ClientCertError,
             ImportError,
+            OSError,
             OpenSSL.crypto.Error,
         ) as caught_exc:
+            self._is_mtls = False
             new_exc = exceptions.MutualTLSChannelError(caught_exc)
             raise new_exc from caught_exc
 

packages/google-auth/tests/test_agent_identity_utils.py

@@ -48,6 +48,17 @@
 
 
 class TestAgentIdentityUtils:
+    @pytest.fixture(autouse=True)
+    def clean_env(self, monkeypatch):
+        monkeypatch.delenv(
+            environment_vars.GOOGLE_API_USE_CLIENT_CERTIFICATE,
+            raising=False,
+        )
+        monkeypatch.delenv(
+            environment_vars.CLOUDSDK_CONTEXT_AWARE_USE_CLIENT_CERTIFICATE,
+            raising=False,
+        )
+
     @mock.patch("cryptography.x509.load_pem_x509_certificate")
     def test_parse_certificate(self, mock_load_cert):
         result = _agent_identity_utils.parse_certificate(b"cert_bytes")
@@ -150,6 +161,33 @@ def test_should_request_bound_token(self, mock_is_agent, monkeypatch):
         )
         assert not _agent_identity_utils.should_request_bound_token(mock.sentinel.cert)
 
+    @mock.patch("google.auth._agent_identity_utils._is_agent_identity_certificate")
+    def test_should_request_bound_token_explicit_use_client_cert_false(
+        self, mock_is_agent, monkeypatch
+    ):
+        mock_is_agent.return_value = True
+        monkeypatch.setenv(
+            environment_vars.GOOGLE_API_USE_CLIENT_CERTIFICATE,
+            "false",
+        )
+        assert not _agent_identity_utils.should_request_bound_token(mock.sentinel.cert)
+
+    @mock.patch("google.auth._agent_identity_utils._is_agent_identity_certificate")
+    def test_should_request_bound_token_explicit_use_client_cert_invalid(
+        self, mock_is_agent, monkeypatch
+    ):
+        mock_is_agent.return_value = True
+        monkeypatch.setenv(
+            environment_vars.GOOGLE_API_USE_CLIENT_CERTIFICATE,
+            "foo",
+        )
+        assert not _agent_identity_utils.should_request_bound_token(mock.sentinel.cert)
+
+    @mock.patch("google.auth._agent_identity_utils._is_agent_identity_certificate")
+    def test_should_request_bound_token_auto_enablement(self, mock_is_agent):
+        mock_is_agent.return_value = True
+        assert _agent_identity_utils.should_request_bound_token(mock.sentinel.cert)
+
     def test_get_agent_identity_certificate_path_success(self, tmpdir, monkeypatch):
         cert_path = tmpdir.join("cert.pem")
         cert_path.write("cert_content")
@@ -439,6 +477,30 @@ def test_get_and_parse_agent_identity_certificate_success(
         mock_parse_certificate.assert_called_once_with(b"cert_bytes")
         assert result == mock_parse_certificate.return_value
 
+    @mock.patch("google.auth._agent_identity_utils.get_agent_identity_certificate_path")
+    def test_get_and_parse_agent_identity_certificate_use_client_cert_false(
+        self, mock_get_path, monkeypatch
+    ):
+        monkeypatch.setenv(
+            environment_vars.GOOGLE_API_USE_CLIENT_CERTIFICATE,
+            "false",
+        )
+        result = _agent_identity_utils.get_and_parse_agent_identity_certificate()
+        assert result is None
+        mock_get_path.assert_not_called()
+
+    @mock.patch("google.auth._agent_identity_utils.get_agent_identity_certificate_path")
+    def test_get_and_parse_agent_identity_certificate_use_client_cert_invalid(
+        self, mock_get_path, monkeypatch
+    ):
+        monkeypatch.setenv(
+            environment_vars.GOOGLE_API_USE_CLIENT_CERTIFICATE,
+            "foo",
+        )
+        result = _agent_identity_utils.get_and_parse_agent_identity_certificate()
+        assert result is None
+        mock_get_path.assert_not_called()
+
     def test_get_cached_cert_fingerprint_no_cert(self):
         with pytest.raises(ValueError, match="mTLS connection is not configured."):
             _agent_identity_utils.get_cached_cert_fingerprint(None)