fix: gracefully fallback if workload fields are missing from cert config (#16022)

Prevents exceptions during gECC flows or when falling back to
SecureConnect by returning None instead of raising ClientCertError when
X.509 workload fields are absent.

---------

Co-authored-by: Daniel Sanche <d.sanche14@gmail.com>

Author: nbayati

packages/google-auth/google/auth/transport/_mtls_helper.py

@@ -191,28 +191,21 @@ def _get_workload_cert_and_key_paths(config_path, include_context_aware=True):
         )
     cert_configs = data["cert_configs"]
 
+    # We return None, None if the expected workload fields are not present.
+    # The certificate config might be present for other types of connections (e.g. gECC),
+    # and we want to gracefully fallback to testing other mTLS configurations
+    # like SecureConnect instead of throwing an exception.
+
     if "workload" not in cert_configs:
-        raise exceptions.ClientCertError(
-            'Certificate config file {} is in an invalid format, a "workload" cert config is expected'.format(
-                absolute_path
-            )
-        )
+        return None, None
     workload = cert_configs["workload"]
 
     if "cert_path" not in workload:
-        raise exceptions.ClientCertError(
-            'Certificate config file {} is in an invalid format, a "cert_path" is expected in the workload cert config'.format(
-                absolute_path
-            )
-        )
+        return None, None
     cert_path = workload["cert_path"]
 
     if "key_path" not in workload:
-        raise exceptions.ClientCertError(
-            'Certificate config file {} is in an invalid format, a "key_path" is expected in the workload cert config'.format(
-                absolute_path
-            )
-        )
+        return None, None
     key_path = workload["key_path"]
 
     # == BEGIN Temporary Cloud Run PATCH ==

packages/google-auth/tests/transport/aio/test_sessions_mtls.py

@@ -98,6 +98,25 @@ async def test_configure_mtls_channel_invalid_format(self):
             with pytest.raises(exceptions.MutualTLSChannelError):
                 await session.configure_mtls_channel()
 
+    @pytest.mark.asyncio
+    async def test_configure_mtls_channel_invalud_fields(self):
+        """
+        If cert is missing expected keys, it should fail gracefully
+        """
+        with mock.patch.dict(
+            os.environ, {"GOOGLE_API_USE_CLIENT_CERTIFICATE": "true"}
+        ), mock.patch("os.path.exists") as mock_exists, mock.patch(
+            "builtins.open", mock.mock_open(read_data='{"cert_configs": {}}')
+        ):
+            mock_exists.return_value = True
+            mock_creds = mock.AsyncMock(spec=credentials.Credentials)
+            session = sessions.AsyncAuthorizedSession(mock_creds)
+
+            await session.configure_mtls_channel()
+
+            # If the file couldn't be parsed, it shouldn't error; it just won't use mTLS
+            assert session._is_mtls is False
+
     @pytest.mark.asyncio
     async def test_configure_mtls_channel_mock_callback(self):
         """

packages/google-auth/tests/transport/test__mtls_helper.py

@@ -577,8 +577,9 @@ def test_no_workload(self, mock_get_cert_config_path, mock_load_json_file):
         mock_get_cert_config_path.return_value = "/path/to/cert"
         mock_load_json_file.return_value = {"cert_configs": {}}
 
-        with pytest.raises(exceptions.ClientCertError):
-            _mtls_helper._get_workload_cert_and_key("")
+        actual_cert, actual_key = _mtls_helper._get_workload_cert_and_key("")
+        assert actual_cert is None
+        assert actual_key is None
 
     @mock.patch("google.auth.transport._mtls_helper._load_json_file", autospec=True)
     @mock.patch(
@@ -590,8 +591,9 @@ def test_no_cert_file(self, mock_get_cert_config_path, mock_load_json_file):
             "cert_configs": {"workload": {"key_path": "path/to/key"}}
         }
 
-        with pytest.raises(exceptions.ClientCertError):
-            _mtls_helper._get_workload_cert_and_key("")
+        actual_cert, actual_key = _mtls_helper._get_workload_cert_and_key("")
+        assert actual_cert is None
+        assert actual_key is None
 
     @mock.patch("google.auth.transport._mtls_helper._load_json_file", autospec=True)
     @mock.patch(
@@ -603,8 +605,9 @@ def test_no_key_file(self, mock_get_cert_config_path, mock_load_json_file):
             "cert_configs": {"workload": {"cert_path": "path/to/key"}}
         }
 
-        with pytest.raises(exceptions.ClientCertError):
-            _mtls_helper._get_workload_cert_and_key("")
+        actual_cert, actual_key = _mtls_helper._get_workload_cert_and_key("")
+        assert actual_cert is None
+        assert actual_key is None
 
 
 class TestReadCertAndKeyFile(object):