feat: Add public wrapper for _mtls_helper.check_use_client_cert which enables mTLS if GOOGLE_API_USE_CLIENT_CERTIFICATE is not set, when the MWID/X.509 cert sources detected (#1859)

Add public wrapper for check_use_client_cert which enables mTLS if
GOOGLE_API_USE_CLIENT_CERTIFICATE is not set, when the MWID/X.509 cert
sources detected. Also, fix check_use_client_cert to return boolean
value.

Change #1848 added the check_use_client_cert method that helps know if
client cert should be used for mTLS connection. However, that was in a
private class, thus, created a public wrapper of the same function so
that it can be used by python Client Libraries. Also, updated
check_use_client_cert to return a boolean value instead of existing
string value for better readability and future scope.

---------

Signed-off-by: Radhika Agrawal <agrawalradhika@google.com>
Co-authored-by: Daniel Sanche <d.sanche14@gmail.com>

Author: agrawalradhika-cell

google/auth/transport/_mtls_helper.py

@@ -408,28 +408,23 @@ def client_cert_callback():
 
 
 def check_use_client_cert():
-    """Returns the value of the GOOGLE_API_USE_CLIENT_CERTIFICATE variable,
-    or an inferred value('true' or 'false') if unset.
+    """Returns boolean for whether the client certificate should be used for mTLS.
 
-    This value is meant to be interpreted as a "true" or "false" value
-    representing whether the client certificate should be used, but could be any
-    arbitrary string.
-
-    If GOOGLE_API_USE_CLIENT_CERTIFICATE is unset, the value value will be
-    inferred by reading a file pointed at by GOOGLE_API_CERTIFICATE_CONFIG, and
-    verifying it contains a "workload" section. If so, the function will return
-    "true", otherwise "false".
+    If GOOGLE_API_USE_CLIENT_CERTIFICATE is set to true or false, a corresponding
+    bool value will be returned. If the value is set to an unexpected string, it
+    will default to False.
+    If GOOGLE_API_USE_CLIENT_CERTIFICATE is unset, the value will be inferred
+    by reading a file pointed at by GOOGLE_API_CERTIFICATE_CONFIG, and verifying
+    it contains a "workload" section. If so, the function will return True,
+    otherwise False.
 
     Returns:
-        str: The value of GOOGLE_API_USE_CLIENT_CERTIFICATE, or an inferred value
-        ("true" or "false") if unset. This string should contain a value, but may
-        be an any arbitrary string read from the user's set
-        GOOGLE_API_USE_CLIENT_CERTIFICATE.
+        bool: Whether the client certificate should be used for mTLS connection.
     """
     use_client_cert = getenv("GOOGLE_API_USE_CLIENT_CERTIFICATE")
     # Check if the value of GOOGLE_API_USE_CLIENT_CERTIFICATE is set.
     if use_client_cert:
-        return use_client_cert.lower()
+        return use_client_cert.lower() == "true"
     else:
         # Check if the value of GOOGLE_API_CERTIFICATE_CONFIG is set.
         cert_path = getenv("GOOGLE_API_CERTIFICATE_CONFIG")
@@ -439,7 +434,7 @@ def check_use_client_cert():
                     content = json.load(f)
                     # verify json has workload key
                     content["cert_configs"]["workload"]
-                    return "true"
+                    return True
             except (
                 FileNotFoundError,
                 OSError,
@@ -448,4 +443,4 @@ def check_use_client_cert():
                 json.JSONDecodeError,
             ) as e:
                 _LOGGER.debug("error decoding certificate: %s", e)
-        return "false"
+        return False

google/auth/transport/grpc.py

@@ -255,13 +255,13 @@ 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 = _mtls_helper.check_use_client_cert()
-        if use_client_cert == "true" and client_cert_callback:
+        if use_client_cert and client_cert_callback:
             # Use the callback if provided.
             cert, key = client_cert_callback()
             ssl_credentials = grpc.ssl_channel_credentials(
                 certificate_chain=cert, private_key=key
             )
-        elif use_client_cert == "true":
+        elif use_client_cert:
             # Use application default SSL credentials.
             adc_ssl_credentils = SslCredentials()
             ssl_credentials = adc_ssl_credentils.ssl_credentials
@@ -292,7 +292,7 @@ class SslCredentials:
 
     def __init__(self):
         use_client_cert = _mtls_helper.check_use_client_cert()
-        if use_client_cert != "true":
+        if not use_client_cert:
             self._is_mtls = False
         else:
             # Load client SSL credentials.

google/auth/transport/mtls.py

@@ -110,3 +110,20 @@ def callback():
         return cert_path, key_path, passphrase_bytes
 
     return callback
+
+
+def should_use_client_cert():
+    """Returns boolean for whether the client certificate should be used for mTLS.
+
+    This is a wrapper around _mtls_helper.check_use_client_cert().
+    If GOOGLE_API_USE_CLIENT_CERTIFICATE is set to true or false, a corresponding
+    bool value will be returned
+    If GOOGLE_API_USE_CLIENT_CERTIFICATE is unset, the value will be inferred by
+    reading a file pointed at by GOOGLE_API_CERTIFICATE_CONFIG, and verifying it
+    contains a "workload" section. If so, the function will return True,
+    otherwise False.
+
+    Returns:
+       bool: indicating whether the client certificate should be used for mTLS.
+    """
+    return _mtls_helper.check_use_client_cert()

google/auth/transport/requests.py

@@ -443,7 +443,7 @@ def configure_mtls_channel(self, client_cert_callback=None):
                 creation failed for any reason.
         """
         use_client_cert = google.auth.transport._mtls_helper.check_use_client_cert()
-        if use_client_cert != "true":
+        if not use_client_cert:
             self._is_mtls = False
             return
         try:

google/auth/transport/urllib3.py

@@ -334,7 +334,7 @@ def configure_mtls_channel(self, client_cert_callback=None):
                 creation failed for any reason.
         """
         use_client_cert = transport._mtls_helper.check_use_client_cert()
-        if use_client_cert != "true":
+        if not use_client_cert:
             return False
         try:
             import OpenSSL

tests/transport/test__mtls_helper.py

@@ -643,7 +643,7 @@ def test_crypto_error(self):
     def test_check_use_client_cert(self, monkeypatch):
         monkeypatch.setenv("GOOGLE_API_USE_CLIENT_CERTIFICATE", "true")
         use_client_cert = _mtls_helper.check_use_client_cert()
-        assert use_client_cert == "true"
+        assert use_client_cert is True
 
     def test_check_use_client_cert_for_workload_with_config_file(self, monkeypatch):
         config_data = {
@@ -663,19 +663,24 @@ def test_check_use_client_cert_for_workload_with_config_file(self, monkeypatch):
         mock_file_handle = mock.mock_open(read_data=config_file_content)
         with mock.patch("builtins.open", mock_file_handle):
             use_client_cert = _mtls_helper.check_use_client_cert()
-            assert use_client_cert == "true"
+            assert use_client_cert is True
 
     def test_check_use_client_cert_false(self, monkeypatch):
         monkeypatch.setenv("GOOGLE_API_USE_CLIENT_CERTIFICATE", "false")
         use_client_cert = _mtls_helper.check_use_client_cert()
-        assert use_client_cert == "false"
+        assert use_client_cert is False
+
+    def test_check_use_client_cert_unsupported_value(self, monkeypatch):
+        monkeypatch.setenv("GOOGLE_API_USE_CLIENT_CERTIFICATE", "dummy")
+        use_client_cert = _mtls_helper.check_use_client_cert()
+        assert use_client_cert is False
 
     def test_check_use_client_cert_for_workload_with_config_file_not_found(
         self, monkeypatch
     ):
         monkeypatch.setenv("GOOGLE_API_USE_CLIENT_CERTIFICATE", "")
         use_client_cert = _mtls_helper.check_use_client_cert()
-        assert use_client_cert == "false"
+        assert use_client_cert is False
 
     def test_check_use_client_cert_for_workload_with_config_file_not_json(
         self, monkeypatch
@@ -688,7 +693,7 @@ def test_check_use_client_cert_for_workload_with_config_file_not_json(
         mock_file_handle = mock.mock_open(read_data=config_file_content)
         with mock.patch("builtins.open", mock_file_handle):
             use_client_cert = _mtls_helper.check_use_client_cert()
-            assert use_client_cert == "false"
+            assert use_client_cert is False
 
     def test_check_use_client_cert_for_workload_with_config_file_no_workload(
         self, monkeypatch
@@ -702,11 +707,11 @@ def test_check_use_client_cert_for_workload_with_config_file_no_workload(
         mock_file_handle = mock.mock_open(read_data=config_file_content)
         with mock.patch("builtins.open", mock_file_handle):
             use_client_cert = _mtls_helper.check_use_client_cert()
-            assert use_client_cert == "false"
+            assert use_client_cert is False
 
     def test_check_use_client_cert_when_file_does_not_exist(self, monkeypatch):
         config_filename = "mock_certificate_config.json"
         monkeypatch.setenv("GOOGLE_API_CERTIFICATE_CONFIG", config_filename)
         monkeypatch.setenv("GOOGLE_API_USE_CLIENT_CERTIFICATE", "")
         use_client_cert = _mtls_helper.check_use_client_cert()
-        assert use_client_cert == "false"
+        assert use_client_cert is False

tests/transport/test_mtls.py

@@ -94,3 +94,12 @@ def test_default_client_encrypted_cert_source(
     callback = mtls.default_client_encrypted_cert_source("cert_path", "key_path")
     with pytest.raises(exceptions.MutualTLSChannelError):
         callback()
+
+
+@mock.patch("google.auth.transport._mtls_helper.check_use_client_cert", autospec=True)
+def test_should_use_client_cert(check_use_client_cert):
+    check_use_client_cert.return_value = mock.Mock()
+    assert mtls.should_use_client_cert()
+
+    check_use_client_cert.return_value = False
+    assert not mtls.should_use_client_cert()