feat: Auto enable mTLS when supported certificates are detected (#1637)

The Python SDK will use a hybrid approach for mTLS enablement:

If the GOOGLE_API_USE_CLIENT_CERTIFICATE environment variable is set
(either true or false or any value), the SDK will respect that setting.
This is necessary for test scenarios and users who need to explicitly
control mTLS behavior.
If the GOOGLE_API_USE_CLIENT_CERTIFICATE environment variable is not
set, the SDK will automatically enable mTLS only if it detects Managed
Workload Identity (MWID) or X.509 Workforce Identity Federation (WIF)
certificate sources. In other cases where the variable is not set, mTLS
will remain disabled.

---------

Signed-off-by: Radhika Agrawal <[email protected]>
Co-authored-by: Chandra Shekhar Sirimala <[email protected]>

Author: agrawalradhika-cell

google/cloud/_storage_v2/services/storage/client.py

@@ -184,6 +184,34 @@ def _get_default_mtls_endpoint(api_endpoint):
     _DEFAULT_ENDPOINT_TEMPLATE = "storage.{UNIVERSE_DOMAIN}"
     _DEFAULT_UNIVERSE = "googleapis.com"
 
+    @staticmethod
+    def _use_client_cert_effective():
+        """Returns whether client certificate should be used for mTLS if the
+        google-auth version supports should_use_client_cert automatic mTLS enablement.
+
+        Alternatively, read from the GOOGLE_API_USE_CLIENT_CERTIFICATE env var.
+
+        Returns:
+            bool: whether client certificate should be used for mTLS
+        Raises:
+            ValueError: (If using a version of google-auth without should_use_client_cert and
+            GOOGLE_API_USE_CLIENT_CERTIFICATE is set to an unexpected value.)
+        """
+        # check if google-auth version supports should_use_client_cert for automatic mTLS enablement
+        if hasattr(mtls, "should_use_client_cert"):
+            return mtls.should_use_client_cert()
+        else:
+            # if unsupported, fallback to reading from env var
+            use_client_cert_str = os.getenv(
+                "GOOGLE_API_USE_CLIENT_CERTIFICATE", "false"
+            ).lower()
+            if use_client_cert_str not in ("true", "false"):
+                raise ValueError(
+                    "Environment variable `GOOGLE_API_USE_CLIENT_CERTIFICATE` must be"
+                    " either `true` or `false`"
+                )
+            return use_client_cert_str == "true"
+
     @classmethod
     def from_service_account_info(cls, info: dict, *args, **kwargs):
         """Creates an instance of this client using the provided credentials
@@ -390,20 +418,16 @@ def get_mtls_endpoint_and_cert_source(
         )
         if client_options is None:
             client_options = client_options_lib.ClientOptions()
-        use_client_cert = os.getenv("GOOGLE_API_USE_CLIENT_CERTIFICATE", "false")
+        use_client_cert = StorageClient._use_client_cert_effective()
         use_mtls_endpoint = os.getenv("GOOGLE_API_USE_MTLS_ENDPOINT", "auto")
-        if use_client_cert not in ("true", "false"):
-            raise ValueError(
-                "Environment variable `GOOGLE_API_USE_CLIENT_CERTIFICATE` must be either `true` or `false`"
-            )
         if use_mtls_endpoint not in ("auto", "never", "always"):
             raise MutualTLSChannelError(
                 "Environment variable `GOOGLE_API_USE_MTLS_ENDPOINT` must be `never`, `auto` or `always`"
             )
 
         # Figure out the client cert source to use.
         client_cert_source = None
-        if use_client_cert == "true":
+        if use_client_cert:
             if client_options.client_cert_source:
                 client_cert_source = client_options.client_cert_source
             elif mtls.has_default_client_cert_source():
@@ -435,20 +459,14 @@ def _read_environment_variables():
             google.auth.exceptions.MutualTLSChannelError: If GOOGLE_API_USE_MTLS_ENDPOINT
                 is not any of ["auto", "never", "always"].
         """
-        use_client_cert = os.getenv(
-            "GOOGLE_API_USE_CLIENT_CERTIFICATE", "false"
-        ).lower()
+        use_client_cert = StorageClient._use_client_cert_effective()
         use_mtls_endpoint = os.getenv("GOOGLE_API_USE_MTLS_ENDPOINT", "auto").lower()
         universe_domain_env = os.getenv("GOOGLE_CLOUD_UNIVERSE_DOMAIN")
-        if use_client_cert not in ("true", "false"):
-            raise ValueError(
-                "Environment variable `GOOGLE_API_USE_CLIENT_CERTIFICATE` must be either `true` or `false`"
-            )
         if use_mtls_endpoint not in ("auto", "never", "always"):
             raise MutualTLSChannelError(
                 "Environment variable `GOOGLE_API_USE_MTLS_ENDPOINT` must be `never`, `auto` or `always`"
             )
-        return use_client_cert == "true", use_mtls_endpoint, universe_domain_env
+        return use_client_cert, use_mtls_endpoint, universe_domain_env
 
     @staticmethod
     def _get_client_cert_source(provided_cert_source, use_cert_flag):

google/cloud/storage/_helpers.py

@@ -111,10 +111,6 @@ def _virtual_hosted_style_base_url(url, bucket, trailing_slash=False):
     return base_url
 
 
-def _use_client_cert():
-    return os.getenv("GOOGLE_API_USE_CLIENT_CERTIFICATE") == "true"
-
-
 def _get_environ_project():
     return os.getenv(
         environment_vars.PROJECT,

google/cloud/storage/client.py

@@ -20,11 +20,12 @@
 import datetime
 import functools
 import json
+import os
 import warnings
 import google.api_core.client_options
 
 from google.auth.credentials import AnonymousCredentials
-
+from google.auth.transport import mtls
 from google.api_core import page_iterator
 from google.cloud._helpers import _LocalStack
 from google.cloud.client import ClientWithProject
@@ -35,7 +36,6 @@
 from google.cloud.storage._helpers import _get_api_endpoint_override
 from google.cloud.storage._helpers import _get_environ_project
 from google.cloud.storage._helpers import _get_storage_emulator_override
-from google.cloud.storage._helpers import _use_client_cert
 from google.cloud.storage._helpers import _virtual_hosted_style_base_url
 from google.cloud.storage._helpers import _DEFAULT_UNIVERSE_DOMAIN
 from google.cloud.storage._helpers import _DEFAULT_SCHEME
@@ -218,7 +218,15 @@ def __init__(
             # The final decision of whether to use mTLS takes place in
             # google-auth-library-python. We peek at the environment variable
             # here only to issue an exception in case of a conflict.
-            if _use_client_cert():
+            use_client_cert = False
+            if hasattr(mtls, "should_use_client_cert"):
+                use_client_cert = mtls.should_use_client_cert()
+            else:
+                use_client_cert = (
+                    os.getenv("GOOGLE_API_USE_CLIENT_CERTIFICATE") == "true"
+                )
+
+            if use_client_cert:
                 raise ValueError(
                     'The "GOOGLE_API_USE_CLIENT_CERTIFICATE" env variable is '
                     'set to "true" and a non-default universe domain is '