feat: Add helper methods for asynchronous x.509 certificate discovery (#1956)

This PR introduces the google.auth.aio.transport.mtls module, providing
asynchronous helper methods for mTLS certificate discovery
[go/caa:x509-async-support](http://goto.google.com/caa:x509-async-support).
These helpers are designed to be for x.509 certs discovery and
non-blocking, ensuring that disk I/O operations are async during mTLS
handshake process. Plus, added unit tests respectively for the helper
functions.
Please note: Only x.509 creds are in scope of this project currently.
Context aware or ECP credentials are not in scope of this project
currently.
Next Steps: Will create a followup PR will that will utilize these
helpers to implement `configure_mtls_channel` within the
`AsyncAuthorizedSession` class.
---------
Signed-off-by: Radhika Agrawal <agrawalradhika@google.com>

Author: agrawalradhika-cell

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

@@ -0,0 +1,137 @@
+# Copyright 2026 Google LLC
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+"""
+Helper functions for mTLS in async for discovery of certs.
+"""
+
+importasyncio
+importlogging
+
+fromgoogle.authimportexceptions
+importgoogle.auth.transport._mtls_helper
+importgoogle.auth.transport.mtls
+
+_LOGGER=logging.getLogger(__name__)
+
+
+asyncdef_run_in_executor(func, *args):
+"""Run a blocking function in an executor to avoid blocking the event loop.
+
+ This implements the non-blocking execution strategy for disk I/O operations.
+ """
+try:
+# For python versions 3.9 and newer versions
+returnawaitasyncio.to_thread(func, *args)
+exceptAttributeError:
+# Fallback for older Python versions
+loop=asyncio.get_running_loop()
+returnawaitloop.run_in_executor(None, func, *args)
+
+
+defdefault_client_cert_source():
+"""Get a callback which returns the default client SSL credentials.
+
+ Returns:
+ Awaitable[Callable[[], [bytes, bytes]]]: A callback which returns the default
+ client certificate bytes and private key bytes, both in PEM format.
+
+ Raises:
+ google.auth.exceptions.DefaultClientCertSourceError: If the default
+ client SSL credentials don't exist or are malformed.
+ """
+ifnotgoogle.auth.transport.mtls.has_default_client_cert_source(
+include_context_aware=False
+ ):
+raiseexceptions.MutualTLSChannelError(
+"Default client cert source doesn't exist"
+ )
+
+asyncdefcallback():
+try:
+_, cert_bytes, key_bytes=awaitget_client_cert_and_key()
+except (OSError, RuntimeError, ValueError) ascaught_exc:
+new_exc=exceptions.MutualTLSChannelError(caught_exc)
+raisenew_excfromcaught_exc
+
+returncert_bytes, key_bytes
+
+returncallback
+
+
+asyncdefget_client_ssl_credentials(
+certificate_config_path=None,
+):
+"""Returns the client side certificate, private key and passphrase.
+
+ We look for certificates and keys with the following order of priority:
+ 1. Certificate and key specified by certificate_config.json.
+ Currently, only X.509 workload certificates are supported.
+
+ Args:
+ certificate_config_path (str): The certificate_config.json file path.
+
+ Returns:
+ Tuple[bool, bytes, bytes, bytes]:
+ A boolean indicating if cert, key and passphrase are obtained, the
+ cert bytes and key bytes both in PEM format, and passphrase bytes.
+
+ Raises:
+ google.auth.exceptions.ClientCertError: if problems occurs when getting
+ the cert, key and passphrase.
+ """
+
+# Attempt to retrieve X.509 Workload cert and key.
+cert, key=await_run_in_executor(
+google.auth.transport._mtls_helper._get_workload_cert_and_key,
+certificate_config_path,
+False,
+ )
+
+ifcertandkey:
+returnTrue, cert, key, None
+
+returnFalse, None, None, None
+
+
+asyncdefget_client_cert_and_key(client_cert_callback=None):
+"""Returns the client side certificate and private key. The function first
+ tries to get certificate and key from client_cert_callback; if the callback
+ is None or doesn't provide certificate and key, the function tries application
+ default SSL credentials.
+
+ Args:
+ client_cert_callback (Optional[Callable[[], (bytes, bytes)]]): An
+ optional callback which returns client certificate bytes and private
+ key bytes both in PEM format.
+
+ Returns:
+ Tuple[bool, bytes, bytes]:
+ A boolean indicating if cert and key are obtained, the cert bytes
+ and key bytes both in PEM format.
+
+ Raises:
+ google.auth.exceptions.ClientCertError: if problems occurs when getting
+ the cert and key.
+ """
+ifclient_cert_callback:
+result=client_cert_callback()
+try:
+cert, key=awaitresult
+exceptTypeError:
+cert, key=result
+returnTrue, cert, key
+
+has_cert, cert, key, _=awaitget_client_ssl_credentials()
+returnhas_cert, cert, key

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

@@ -25,6 +25,8 @@
 fromgoogle.authimportexceptions
 
 CONTEXT_AWARE_METADATA_PATH="~/.secureConnect/context_aware_metadata.json"
+
+# Default gcloud config path, to be used with path.expanduser for cross-platform compatibility.
 CERTIFICATE_CONFIGURATION_DEFAULT_PATH="~/.config/gcloud/certificate_config.json"
 _CERT_PROVIDER_COMMAND="cert_provider_command"
 _CERT_REGEX=re.compile(
@@ -103,14 +105,18 @@ def _load_json_file(path):
 returnjson_data
 
 
-def_get_workload_cert_and_key(certificate_config_path=None):
+def_get_workload_cert_and_key(
+certificate_config_path=None, include_context_aware=True
+):
 """Read the workload identity cert and key files specified in the certificate config provided.
  If no config path is provided, check the environment variable: "GOOGLE_API_CERTIFICATE_CONFIG"
  first, then the well known gcloud location: "~/.config/gcloud/certificate_config.json".
 
  Args:
  certificate_config_path (string): The certificate config path. If no path is provided,
  the environment variable will be checked first, then the well known gcloud location.
+ include_context_aware (bool): If context aware metadata path should be checked for the
+ SecureConnect mTLS configuration.
 
  Returns:
  Tuple[Optional[bytes], Optional[bytes]]: client certificate bytes in PEM format and key
@@ -121,15 +127,17 @@ def _get_workload_cert_and_key(certificate_config_path=None):
  the certificate or key information.
  """
 
-cert_path, key_path=_get_workload_cert_and_key_paths(certificate_config_path)
+cert_path, key_path=_get_workload_cert_and_key_paths(
+certificate_config_path, include_context_aware
+ )
 
 ifcert_pathisNoneandkey_pathisNone:
 returnNone, None
 
 return_read_cert_and_key_files(cert_path, key_path)
 
 
-def_get_cert_config_path(certificate_config_path=None):
+def_get_cert_config_path(certificate_config_path=None, include_context_aware=True):
 """Get the certificate configuration path based on the following order:
 
  1: Explicit override, if set
@@ -141,6 +149,8 @@ def _get_cert_config_path(certificate_config_path=None):
  Args:
  certificate_config_path (string): The certificate config path. If provided, the well known
  location and environment variable will be ignored.
+ include_context_aware (bool): If context aware metadata path should be checked for the
+ SecureConnect mTLS configuration.
 
  Returns:
  The absolute path of the certificate config file, and None if the file does not exist.
@@ -155,7 +165,7 @@ def _get_cert_config_path(certificate_config_path=None):
 environment_vars.CLOUDSDK_CONTEXT_AWARE_CERTIFICATE_CONFIG_FILE_PATH,
 None,
  )
-ifenv_pathisnotNoneandenv_path!="":
+ifinclude_context_awareandenv_pathisnotNoneandenv_path!="":
 certificate_config_path=env_path
 else:
 certificate_config_path=CERTIFICATE_CONFIGURATION_DEFAULT_PATH
@@ -166,8 +176,8 @@ def _get_cert_config_path(certificate_config_path=None):
 returncertificate_config_path
 
 
-def_get_workload_cert_and_key_paths(config_path):
-absolute_path=_get_cert_config_path(config_path)
+def_get_workload_cert_and_key_paths(config_path, include_context_aware=True):
+absolute_path=_get_cert_config_path(config_path, include_context_aware)
 ifabsolute_pathisNone:
 returnNone, None
 

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

@@ -20,14 +20,19 @@
 fromgoogle.auth.transportimport_mtls_helper
 
 
-defhas_default_client_cert_source():
+defhas_default_client_cert_source(include_context_aware=True):
 """Check if default client SSL credentials exists on the device.
 
+ Args:
+ include_context_aware (bool): include_context_aware indicates if context_aware
+ path location will be checked or should it be skipped.
+
  Returns:
  bool: indicating if the default client cert source exists.
  """
 if (
-_mtls_helper._check_config_path(_mtls_helper.CONTEXT_AWARE_METADATA_PATH)
+include_context_aware
+and_mtls_helper._check_config_path(_mtls_helper.CONTEXT_AWARE_METADATA_PATH)
 isnotNone
  ):
 returnTrue
@@ -58,7 +63,7 @@ def default_client_cert_source():
  google.auth.exceptions.DefaultClientCertSourceError: If the default
  client SSL credentials don't exist or are malformed.
  """
-ifnothas_default_client_cert_source():
+ifnothas_default_client_cert_source(include_context_aware=True):
 raiseexceptions.MutualTLSChannelError(
 "Default client cert source doesn't exist"
  )
@@ -94,7 +99,7 @@ def default_client_encrypted_cert_source(cert_path, key_path):
  google.auth.exceptions.DefaultClientCertSourceError: If any problem
  occurs when loading or saving the client certificate and key.
  """
-ifnothas_default_client_cert_source():
+ifnothas_default_client_cert_source(include_context_aware=True):
 raiseexceptions.MutualTLSChannelError(
 "Default client encrypted cert source doesn't exist"
  )

packages/google-auth/system_tests/system_tests_sync/test_service_account.py

@@ -41,12 +41,12 @@ def test_refresh_success(http_request, credentials, token_info):
 
 assertinfo["email"] ==credentials.service_account_email
 info_scopes=_helpers.string_to_scopes(info["scope"])
-assertset(info_scopes)==set(
+assertset(info_scopes).issubset(set(
  [
 "https://www.googleapis.com/auth/userinfo.email",
 "https://www.googleapis.com/auth/userinfo.profile",
  ]
- )
+ ))
 
 deftest_iam_signer(http_request, credentials):
 credentials=credentials.with_scopes(