feat: Update the helper for use_client_cert and add support for grpc.py

Signed-off-by: Radhika Agrawal <agrawalradhika@google.com>

Author: agrawalradhika-cell

google/auth/transport/_mtls_helper.py

@@ -407,13 +407,42 @@ def client_cert_callback():
     # Then dump the decrypted key bytes
     return crypto.dump_privatekey(crypto.FILETYPE_PEM, pkey)
 
-def check_use_client_cert_for_workload(use_client_cert):
-  """Checks if the workload should use client cert for mutual TLS."""
-  if use_client_cert == "":
+def check_use_client_cert():
+  """Returns the effective value of use_client_cert to be used.
+
+  Returns:
+      str:
+          A boolean indicating if client certificate should be used.
+          The value is "true" or "false" or unset.
+          If unset, the function checks if the GOOGLE_API_CERTIFICATE_CONFIG
+          environment variable is set. If it is set, the function returns
+          "true" if the certificate config file contains "workload" section
+          and "false" otherwise.
+
+  Raises:
+      ValueError: if GOOGLE_API_USE_CLIENT_CERTIFICATE is set to unsupported
+      value, which is not "true" or "false".
+  """
+  use_client_cert = os.getenv("GOOGLE_API_USE_CLIENT_CERTIFICATE")
+  ### Check if the value of GOOGLE_API_USE_CLIENT_CERTIFICATE is unset.
+  if use_client_cert == "" or use_client_cert is None:
     cert_path = os.getenv("GOOGLE_API_CERTIFICATE_CONFIG")
     if cert_path:
       with open(cert_path, "r") as f:
         content = f.read()
         if "workload" in content:
-          return True
-    return False
+          return "true"
+    return "false"
+  else:
+    ### Check if the value of GOOGLE_API_USE_CLIENT_CERTIFICATE is set but to an
+    ### invalid value.
+    use_client_cert = use_client_cert.lower()
+    if use_client_cert not in ("true", "false"):
+      raise ValueError(
+          "Environment variable `GOOGLE_API_USE_CLIENT_CERTIFICATE` must be"
+          " either `true` or `false`"
+      )
+    else:
+    ### Return the value of GOOGLE_API_USE_CLIENT_CERTIFICATE which is set.
+      return use_client_cert
+

google/auth/transport/grpc.py

@@ -256,8 +256,7 @@ def my_client_cert_callback():
 
     # If SSL credentials are not explicitly set, try client_cert_callback and ADC.
     if not ssl_credentials:
-        use_client_cert = os.getenv(
-            environment_vars.GOOGLE_API_USE_CLIENT_CERTIFICATE, "false"
+	use_client_cert = _mtls_helper.check_use_client_cert()
         )
         if use_client_cert == "true" and client_cert_callback:
             # Use the callback if provided.
@@ -295,8 +294,7 @@ class SslCredentials:
     """
 
     def __init__(self):
-        use_client_cert = os.getenv(
-            environment_vars.GOOGLE_API_USE_CLIENT_CERTIFICATE, "false"
+	use_client_cert = _mtls_helper.check_use_client_cert()
         )
         if use_client_cert != "true":
             self._is_mtls = False

google/auth/transport/requests.py

@@ -444,21 +444,10 @@ def configure_mtls_channel(self, client_cert_callback=None):
             google.auth.exceptions.MutualTLSChannelError: If mutual TLS channel
                 creation failed for any reason.
         """
-        use_client_cert = os.getenv(
-            environment_vars.GOOGLE_API_USE_CLIENT_CERTIFICATE)
-        if use_client_cert != "true":
-            ## Checking if the GOOGLE_API_USE_CLIENT_CERTIFICATE is unset.
-            if _mtls_helper.check_use_client_cert_for_workload(
-                use_client_cert
-            ):
-                os.putenv(
-                    environment_vars.GOOGLE_API_USE_CLIENT_CERTIFICATE, "true"
-                )
-                use_client_cert = "true"
-            else:
-                use_client_cert = "false"
-                self._is_mtls = False
-                return
+	use_client_cert = _mtls_helper.check_use_client_cert()
+	if use_client_cert != "true":
+            self._is_mtls = False
+            return
         try:
             import OpenSSL
         except ImportError as caught_exc:

google/auth/transport/urllib3.py

@@ -335,22 +335,9 @@ def configure_mtls_channel(self, client_cert_callback=None):
             google.auth.exceptions.MutualTLSChannelError: If mutual TLS channel
                 creation failed for any reason.
         """
-        use_client_cert = os.getenv(
-            environment_vars.GOOGLE_API_USE_CLIENT_CERTIFICATE)
-        if use_client_cert != "true":
-            ## Check if workload is present in the certificate config file 
-            ## and GOOGLE_API_USE_CLIENT_CERTIFICATE is unset.
-            if _mtls_helper.check_use_client_cert_for_workload(
-                use_client_cert
-            ):
-                os.putenv(
-                    environment_vars.GOOGLE_API_USE_CLIENT_CERTIFICATE, "true"
-                )
-                use_client_cert = "true"
-            else:
-                use_client_cert = "false"
-                return False
-
+	use_client_cert = _mtls_helper.check_use_client_cert()
+	if use_client_cert != "true":
+            return False
         try:
             import OpenSSL
         except ImportError as caught_exc:

tests/transport/test__mtls_helper.py

@@ -640,9 +640,10 @@ def test_crypto_error(self):
                 ENCRYPTED_EC_PRIVATE_KEY, b"wrong_password"
             )
 
-    def test_check_use_client_cert_for_workload(self):
-        use_client_cert = _mtls_helper.check_use_client_cert_for_workload("")
-        assert use_client_cert == False
+    def test_check_use_client_cert(self):
+        os.environ["GOOGLE_API_USE_CLIENT_CERTIFICATE"] = "true"
+        use_client_cert = _mtls_helper.check_use_client_cert()
+        assert use_client_cert == "true"
 
     def test_check_use_client_cert_for_workload_with_config_file(self):
         config_data = {
@@ -660,7 +661,7 @@ def test_check_use_client_cert_for_workload_with_config_file(self):
         m = mock.mock_open(read_data=config_file_content)
         with mock.patch("builtins.open", m):
             os.environ["GOOGLE_API_CERTIFICATE_CONFIG"] = config_filename
-            use_client_cert = _mtls_helper.check_use_client_cert_for_workload(
-                ""
-            )
-            assert use_client_cert == True
+            os.environ["GOOGLE_API_USE_CLIENT_CERTIFICATE"] = ""
+            use_client_cert = _mtls_helper.check_use_client_cert()
+            assert use_client_cert == "true"
+