feat: Auto enable mTLS when supported certificates are detected

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 '