fix(auth): Agentic Identites mTLS gaps fix _is_mtls and SslCredentials. (#17387)

Resolve some mTLS transport state and gRPC workload certificate issues


- Fix inconsistent `_is_mtls` flag in `urllib3` on fallback to default
HTTP.

- Reset `_is_mtls` to `False` in async `sessions` if a custom request
handler cannot be reconfigured.

- Update gRPC `SslCredentials` to recognize workload certificates by
checking `has_default_client_cert_source` instead of only SecureConnect.

Author: vverman

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

@@ -182,18 +182,16 @@ async def _do_configure():
                     google.auth.transport._mtls_helper.check_use_client_cert
                 )
                 if not use_client_cert:
-                    self._is_mtls = False
                     return
 
                 try:
                     (
-                        self._is_mtls,
+                        is_mtls,
                         cert,
                         key,
                     ) = await mtls.get_client_cert_and_key(client_cert_callback)
 
-                    if self._is_mtls:
-                        self._cached_cert = cert
+                    if is_mtls:
                         ssl_context = await mtls._run_in_executor(
                             mtls.make_client_cert_ssl_context, cert, key
                         )
@@ -208,9 +206,13 @@ async def _do_configure():
                             old_auth_request = self._auth_request
                             self._auth_request = AiohttpRequest(session=new_session)
 
-                            await old_auth_request.close()
+                            try:
+                                await old_auth_request.close()
+                            except Exception:
+                                # Suppress so it doesn't abort the mTLS configuration
+                                pass
                         else:
-                            self._is_mtls = False
+                            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. "
@@ -220,12 +222,13 @@ async def _do_configure():
                                 UserWarning,
                             )
 
-                except (
-                    exceptions.ClientCertError,
-                    ImportError,
-                    OSError,
-                ) as caught_exc:
-                    self._is_mtls = False
+                    self._is_mtls = is_mtls
+                    if is_mtls:
+                        self._cached_cert = cert
+                    else:
+                        self._cached_cert = None
+
+                except Exception as caught_exc:
                     new_exc = exceptions.MutualTLSChannelError(caught_exc)
                     raise new_exc from caught_exc
 
@@ -586,4 +589,10 @@ async def close(self) -> None:
         """
         Close the underlying auth request session.
         """
+        if self._mtls_init_task and not self._mtls_init_task.done():
+            self._mtls_init_task.cancel()
+            try:
+                await self._mtls_init_task
+            except asyncio.CancelledError:
+                pass
         await self._auth_request.close()

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

@@ -20,6 +20,7 @@
 
 from google.auth import exceptions
 from google.auth.transport import _mtls_helper
+from google.auth.transport import mtls
 from google.oauth2 import service_account
 
 try:
@@ -279,14 +280,19 @@ def my_client_cert_callback():
 class SslCredentials:
     """Class for application default SSL credentials.
 
-    The behavior is controlled by `GOOGLE_API_USE_CLIENT_CERTIFICATE` environment
-    variable whose default value is `false`. Client certificate will not be used
-    unless the environment variable is explicitly set to `true`. See
-    https://google.aip.dev/auth/4114
+    Mutual TLS (mTLS) is enabled if either:
 
-    If the environment variable is `true`, then for devices with endpoint verification
-    support, a device certificate will be automatically loaded and mutual TLS will
-    be established.
+    1. The `GOOGLE_API_USE_CLIENT_CERTIFICATE` environment variable is explicitly
+       set to `"true"`.
+    2. The `GOOGLE_API_USE_CLIENT_CERTIFICATE` environment variable is unset or empty,
+       but a valid workload certificate configuration is found (e.g., via the
+       `GOOGLE_API_CERTIFICATE_CONFIG` environment variable or the default gcloud config path).
+
+    See https://google.aip.dev/auth/4114 for client certificate discovery details.
+
+    If client certificate usage is enabled, then for devices with endpoint
+    verification support, a device certificate will be automatically loaded and
+    mutual TLS will be established.
     See https://cloud.google.com/endpoint-verification/docs/overview.
     """
 
@@ -295,11 +301,7 @@ def __init__(self):
         if not use_client_cert:
             self._is_mtls = False
         else:
-            # Load client SSL credentials.
-            metadata_path = _mtls_helper._check_config_path(
-                _mtls_helper.CONTEXT_AWARE_METADATA_PATH
-            )
-            self._is_mtls = metadata_path is not None
+            self._is_mtls = mtls.has_default_client_cert_source()
 
     @property
     def ssl_credentials(self):
@@ -319,11 +321,15 @@ def ssl_credentials(self):
         """
         if self._is_mtls:
             try:
-                _, cert, key, _ = _mtls_helper.get_client_ssl_credentials()
-                self._ssl_credentials = grpc.ssl_channel_credentials(
-                    certificate_chain=cert, private_key=key
-                )
-            except exceptions.ClientCertError as caught_exc:
+                has_cert, cert, key, _ = _mtls_helper.get_client_ssl_credentials()
+                if has_cert:
+                    self._ssl_credentials = grpc.ssl_channel_credentials(
+                        certificate_chain=cert, private_key=key
+                    )
+                else:
+                    self._ssl_credentials = grpc.ssl_channel_credentials()
+                    self._is_mtls = False
+            except (exceptions.ClientCertError, OSError) as caught_exc:
                 new_exc = exceptions.MutualTLSChannelError(caught_exc)
                 raise new_exc from caught_exc
         else:

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

@@ -443,42 +443,49 @@ def configure_mtls_channel(self, client_cert_callback=None):
 
         Raises:
             google.auth.exceptions.MutualTLSChannelError: If mutual TLS channel
-                creation failed for any reason.
+                creation failed for any reason. The existing session state (such
+                as adapter mounts) remains unmodified if this error is raised.
         """
         use_client_cert = google.auth.transport._mtls_helper.check_use_client_cert()
         if not use_client_cert:
-            self._is_mtls = False
             return
+
         try:
             import OpenSSL
         except ImportError as caught_exc:
-            self._is_mtls = False
             new_exc = exceptions.MutualTLSChannelError(caught_exc)
             raise new_exc from caught_exc
 
         try:
             (
-                self._is_mtls,
+                is_mtls,
                 cert,
                 key,
             ) = google.auth.transport._mtls_helper.get_client_cert_and_key(
                 client_cert_callback
             )
 
-            if self._is_mtls:
-                mtls_adapter = _MutualTlsAdapter(cert, key)
-                self._cached_cert = cert
-                self.mount("https://", mtls_adapter)
+            if is_mtls:
+                new_adapter = _MutualTlsAdapter(cert, key)
+            else:
+                new_adapter = requests.adapters.HTTPAdapter()
         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
 
+        self.mount("https://", new_adapter)
+        self._is_mtls = is_mtls
+        if is_mtls:
+            self._cached_cert = cert
+        else:
+            if hasattr(self, "_cached_cert"):
+                del self._cached_cert
+
     def request(
         self,
         method,

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

@@ -332,18 +332,16 @@ def configure_mtls_channel(self, client_cert_callback=None):
 
         Raises:
             google.auth.exceptions.MutualTLSChannelError: If mutual TLS channel
-                creation failed for any reason.
+                creation failed for any reason. The existing channel state (the
+                HTTP client) remains unmodified if this error is raised.
         """
         use_client_cert = transport._mtls_helper.check_use_client_cert()
         if not use_client_cert:
-            self._is_mtls = False
             return False
-        else:
-            self._is_mtls = True
+
         try:
             import OpenSSL
         except ImportError as caught_exc:
-            self._is_mtls = False
             new_exc = exceptions.MutualTLSChannelError(caught_exc)
             raise new_exc from caught_exc
 
@@ -353,21 +351,29 @@ def configure_mtls_channel(self, client_cert_callback=None):
             )
 
             if found_cert_key:
-                self.http = _make_mutual_tls_http(cert, key)
-                self._cached_cert = cert
+                new_http = _make_mutual_tls_http(cert, key)
+                new_is_mtls = True
             else:
-                self.http = _make_default_http()
-                self._is_mtls = False
+                new_http = _make_default_http()
+                new_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
 
+        self.http = new_http
+        self._is_mtls = new_is_mtls
+        self._request.http = new_http
+        if new_is_mtls:
+            self._cached_cert = cert
+        else:
+            if hasattr(self, "_cached_cert"):
+                del self._cached_cert
+
         if self._has_user_provided_http:
             self._has_user_provided_http = False
             warnings.warn(