feat: Autoupdate the GOOGLE_API_USE_CLIENT_CERTIFICATE flag to true if not set, if the MWID/X.509 cert sources detected

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

Author: agrawalradhika-cell

google/auth/transport/_mtls_helper.py

@@ -17,6 +17,7 @@
 import json
 import logging
 from os import environ, path
+import os
 import re
 import subprocess
 
@@ -405,3 +406,14 @@ 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 == "":
+    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

google/auth/transport/requests.py

@@ -445,12 +445,20 @@ def configure_mtls_channel(self, client_cert_callback=None):
                 creation failed for any reason.
         """
         use_client_cert = os.getenv(
-            environment_vars.GOOGLE_API_USE_CLIENT_CERTIFICATE, "false"
-        )
+            environment_vars.GOOGLE_API_USE_CLIENT_CERTIFICATE)
         if use_client_cert != "true":
-            self._is_mtls = False
-            return
-
+            ## 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
         try:
             import OpenSSL
         except ImportError as caught_exc:

google/auth/transport/urllib3.py

@@ -336,10 +336,20 @@ def configure_mtls_channel(self, client_cert_callback=None):
                 creation failed for any reason.
         """
         use_client_cert = os.getenv(
-            environment_vars.GOOGLE_API_USE_CLIENT_CERTIFICATE, "false"
-        )
+            environment_vars.GOOGLE_API_USE_CLIENT_CERTIFICATE)
         if use_client_cert != "true":
-            return False
+            ## 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
 
         try:
             import OpenSSL

tests/transport/test__mtls_helper.py

@@ -12,6 +12,7 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
+import json
 import os
 import re
 
@@ -638,3 +639,28 @@ def test_crypto_error(self):
             _mtls_helper.decrypt_private_key(
                 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_for_workload_with_config_file(self):
+        config_data = {
+            "version": 1,
+            "cert_configs": {
+                "workload": {
+                    "cert_path": "path/to/cert/file",
+                    "key_path": "path/to/key/file",
+                }
+            },
+        }
+        config_filename = "mock_certificate_config.json"
+        config_file_content = json.dumps(config_data)
+        # Use mock_open to simulate the file in memory
+        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