fix(auth): prevent TypeError and support home-dir cert fallback for X… (#18016)

….509 WIF on ECP machines

- Prevent TypeError crash in identity_pool.py by raising ClientCertError
if _get_mtls_cert_and_key_paths() returns None for the certificate path.
- Add fallback in _mtls_helper.py to check the default home directory
configuration ~/.config/gcloud/certificate_config.json if the
env-var-resolved config does not contain a workload block.
- Add unit tests to cover both behaviors and verify they function
correctly.

Fixes: b/542359992


## Bug / Context
Fixes the TypeError (NoneType) crash that occurs when developers attempt
X.509 Workload Identity Federation on ECP machines. (b/542359992)

## Changes
- **Fallback to user configuration:** In `_mtls_helper.py`, if the ECP
system-wide config lacks a `workload` block, fallback to check the home
folder's `~/.config/gcloud/certificate_config.json`.
- **TypeError Prevention:** In `identity_pool.py`, raise a clean
`ClientCertError` if no certificate path is configured, preventing a
generic `NoneType` crash in `open()`.
- **Defensive Check:** Defensively assert that loaded JSON config data
is a dictionary before accessing fields.
- **Unit Tests:** Added tests to cover these fallback and error handling
scenarios.

## Verification
You can verify the fix by running this python script with the local
packages loaded:

```python
import os
import sys
import json
from google.auth import identity_pool
from google.auth import exceptions

# Setup dummy config using default cert path config
cred_config = {
    "type": "external_account",
    "audience": "//iam.googleapis.com/projects/123456/locations/global/workloadIdentityPools/test-pool/providers/test-provider",
    "subject_token_type": "urn:ietf:params:oauth:token-type:jwt",
    "token_url": "https://sts.googleapis.com/v1/token",
    "credential_source": {
        "certificate": {
            "use_default_certificate_config": True
        }
    }
}
config_file = "verify_cred_config.json"
with open(config_file, "w") as f:
    json.dump(cred_config, f)

try:
    credentials = identity_pool.Credentials.from_file(config_file)
    cert_bytes = credentials._get_cert_bytes()
    print("Success: Certificate read successfully.")
except exceptions.ClientCertError as e:
    print(f"Verified: Prevented TypeError crash. Raised ClientCertError: {e}")
finally:
    if os.path.exists(config_file):
        os.remove(config_file)

Author: attharva-24

GEMINI.md

@@ -0,0 +1,20 @@
+# Google Cloud Python Workspace Rules
+
+These guidelines are automatically applied to Python development tasks within this repository.
+
+---
+
+## 1. Filesystem and Path Resolution
+* **Dynamic Configuration Directories:** Never hardcode paths like `~/.config/gcloud/` or standard user directories. Always utilize existing SDK helpers (such as `_cloud_sdk.get_config_path()`) to dynamically locate system and configuration files.
+* **Path Normalization:** When comparing path strings (especially paths retrieved from environment variables or dynamically built), always normalize them using `os.path.normpath` or `pathlib.Path` to prevent Windows vs Unix slash mismatch issues (`\` vs `/`).
+
+## 2. Input Validation (Defensive Programming)
+* **Untrusted File Inputs:** Any data loaded from external configuration files (JSON, YAML, CSV) is untrusted. Always type-validate structure (e.g. check `isinstance(data, dict)` and `isinstance(data.get("sub_key"), dict)`) *before* indexing or calling dictionary lookup keys, avoiding `TypeError` exceptions.
+
+## 3. Exception Contract Compliance
+* **Public Interface Contracts:** When introducing new exception pathways in internal helpers, always trace their propagation. If a public-facing API method (e.g. `refresh()`) is documented to raise a specific base exception class (like `RefreshError`), wrap lower-level custom exceptions (like `ClientCertError`) or system exceptions (like `OSError`) and re-raise them under the correct interface exception types.
+* **Self-Contained Fallbacks:** Fallback logic must be resilient and self-contained. Always wrap fallback configuration loading in try-except blocks to catch expected exceptions (like `ClientCertError` or `OSError`) and bypass failures gracefully.
+
+## 4. Unit Testing and Mock Hygiene
+* **Localized Mocking:** When mocking standard functions or filesystem checks (like `path.exists`), mock the local module import path (e.g., `google.auth.transport._mtls_helper.path.exists`) instead of patching builtins globally (e.g., `os.path.exists`), ensuring mocks are isolated.
+* **Fallback Verification:** Fallback test cases must explicitly verify execution flow by asserting the expected call sequence and arguments of mocked helpers using `assert_called_once_with` or `assert_has_calls`.

packages/google-auth/google/auth/identity_pool.py

@@ -412,6 +412,10 @@ def _get_mtls_cert_and_key_paths(self):
 
     def _get_cert_bytes(self):
         cert_path, _ = self._get_mtls_cert_and_key_paths()
+        if cert_path is None:
+            raise exceptions.ClientCertError(
+                "Workload certificate configuration could not be found or does not contain workload certificate paths."
+            )
         return _mtls_helper._read_cert_file(cert_path)
 
     def _mtls_required(self):
@@ -568,7 +572,13 @@ def refresh(self, request):
         cert_fingerprint = None
         # Check if the credential is X.509 based.
         if self._credential_source_certificate is not None:
-            cert_bytes = self._get_cert_bytes()
+            try:
+                cert_bytes = self._get_cert_bytes()
+            except (exceptions.ClientCertError, OSError) as e:
+                raise exceptions.RefreshError(
+                    "Failed to retrieve certificate bytes for external"
+                    " account credentials"
+                ) from e
             cert = _agent_identity_utils.parse_certificate(cert_bytes)
             if _agent_identity_utils.should_request_bound_token(cert):
                 cert_fingerprint = (

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

@@ -459,7 +459,11 @@ def _get_workload_cert_and_key_paths(config_path, include_context_aware=True):
 
     data = _load_json_file(absolute_path)
 
-    if "cert_configs" not in data:
+    if (
+        not isinstance(data, dict)
+        or "cert_configs" not in data
+        or not isinstance(data["cert_configs"], dict)
+    ):
         raise exceptions.ClientCertError(
             'Certificate config file {} is in an invalid format, a "cert configs" object is expected'.format(
                 absolute_path
@@ -472,11 +476,40 @@ def _get_workload_cert_and_key_paths(config_path, include_context_aware=True):
     # and we want to gracefully fallback to testing other mTLS configurations
     # like SecureConnect instead of throwing an exception.
 
-    if "workload" not in cert_configs:
+    if (
+        not isinstance(cert_configs, dict) or "workload" not in cert_configs
+    ) and config_path is None:
+        default_home_path = path.expanduser(
+            os.path.join(
+                _cloud_sdk.get_config_path(),
+                "certificate_config.json",
+            )
+        )
+        if path.exists(default_home_path) and os.path.normpath(
+            default_home_path
+        ) != os.path.normpath(absolute_path):
+            try:
+                home_data = _load_json_file(default_home_path)
+                if isinstance(home_data, dict):
+                    home_cert_configs = home_data.get("cert_configs")
+                    if (
+                        isinstance(home_cert_configs, dict)
+                        and "workload" in home_cert_configs
+                    ):
+                        cert_configs = home_cert_configs
+                        absolute_path = default_home_path
+            except (exceptions.ClientCertError, OSError):
+                pass
+
+    if not isinstance(cert_configs, dict) or "workload" not in cert_configs:
         return None, None
     workload = cert_configs["workload"]
 
-    if "cert_path" not in workload or "key_path" not in workload:
+    if (
+        not isinstance(workload, dict)
+        or "cert_path" not in workload
+        or "key_path" not in workload
+    ):
         raise exceptions.ClientCertError(
             'Workload certificate configuration is missing "cert_path" or "key_path" in {}'.format(
                 absolute_path

packages/google-auth/tests/test_identity_pool.py

@@ -1784,6 +1784,57 @@ def test_get_mtls_certs_invalid(self):
             'The credential is not configured to use mtls requests. The credential should include a "certificate" section in the credential source.'
         )
 
+    @mock.patch(
+        "google.auth.transport._mtls_helper._get_workload_cert_and_key_paths",
+        return_value=(None, None),
+    )
+    def test_get_cert_bytes_none_raises_error(
+        self, mock_get_workload_cert_and_key_paths
+    ):
+        credentials = self.make_credentials(
+            credential_source=self.CREDENTIAL_SOURCE_CERTIFICATE.copy()
+        )
+
+        with pytest.raises(exceptions.ClientCertError) as excinfo:
+            credentials._get_cert_bytes()
+
+        assert excinfo.match(
+            "Workload certificate configuration could not be found or does not contain workload certificate paths."
+        )
+
+    @mock.patch.object(
+        identity_pool.Credentials,
+        "_get_cert_bytes",
+        side_effect=exceptions.ClientCertError("mock error"),
+    )
+    def test_refresh_cert_error_raises_refresh_error(self, mock_get_cert_bytes):
+        credentials = self.make_credentials(
+            credential_source=self.CREDENTIAL_SOURCE_CERTIFICATE.copy()
+        )
+
+        with pytest.raises(exceptions.RefreshError) as excinfo:
+            credentials.refresh(None)
+
+        assert excinfo.match(
+            "Failed to retrieve certificate bytes for external account credentials"
+        )
+
+    @mock.patch.object(
+        identity_pool.Credentials,
+        "_get_cert_bytes",
+        side_effect=OSError("mock os error"),
+    )
+    def test_refresh_os_error_raises_refresh_error(self, mock_get_cert_bytes):
+        credentials = self.make_credentials(
+            credential_source=self.CREDENTIAL_SOURCE_CERTIFICATE.copy()
+        )
+
+        with pytest.raises(exceptions.RefreshError) as excinfo:
+            credentials.refresh(None)
+
+        msg = "Failed to retrieve certificate bytes for external"
+        assert excinfo.match(msg + " account credentials")
+
     @mock.patch("google.auth._agent_identity_utils.parse_certificate")
     @mock.patch(
         "google.auth._agent_identity_utils.should_request_bound_token",

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

@@ -499,6 +499,43 @@ def test_no_cert_configs(
         with pytest.raises(exceptions.ClientCertError):
             _mtls_helper._get_workload_cert_and_key("")
 
+    @mock.patch("google.auth.transport._mtls_helper._load_json_file", autospec=True)
+    @mock.patch("google.auth.transport._mtls_helper.path.exists", autospec=True)
+    def test_non_dict_cert_configs_raises_error(
+        self, mock_path_exists, mock_load_json_file
+    ):
+        mock_path_exists.return_value = True
+
+        for val in [None, [], "not_a_dict"]:
+            mock_load_json_file.return_value = {"cert_configs": val}
+            with pytest.raises(exceptions.ClientCertError):
+                _mtls_helper._get_workload_cert_and_key(None)
+
+    @mock.patch("google.auth.transport._mtls_helper._load_json_file", autospec=True)
+    @mock.patch("google.auth.transport._mtls_helper.path.exists", autospec=True)
+    def test_malformed_json_returns_error(self, mock_path_exists, mock_load_json_file):
+        mock_path_exists.return_value = True
+
+        for val in [None, [], "invalid_string"]:
+            mock_load_json_file.return_value = val
+            with pytest.raises(exceptions.ClientCertError):
+                _mtls_helper._get_workload_cert_and_key(None)
+
+    @mock.patch("google.auth.transport._mtls_helper._load_json_file", autospec=True)
+    @mock.patch("google.auth.transport._mtls_helper.path.exists", autospec=True)
+    def test_non_dict_workload_raises_error(
+        self, mock_path_exists, mock_load_json_file
+    ):
+        mock_path_exists.return_value = True
+
+        for invalid_workload in [None, 123, "not_a_dict"]:
+            mock_load_json_file.return_value = {
+                "cert_configs": {"workload": invalid_workload}
+            }
+
+            with pytest.raises(exceptions.ClientCertError):
+                _mtls_helper._get_workload_cert_and_key(None)
+
     @mock.patch("google.auth.transport._mtls_helper._load_json_file", autospec=True)
     @mock.patch(
         "google.auth.transport._mtls_helper._get_cert_config_path", autospec=True
@@ -511,6 +548,182 @@ def test_no_workload(self, mock_get_cert_config_path, mock_load_json_file):
         assert actual_cert is None
         assert actual_key is None
 
+    @mock.patch(
+        "google.auth.transport._mtls_helper._load_json_file", autospec=True
+    )  # noqa: E501
+    @mock.patch(
+        "google.auth.transport._mtls_helper._get_cert_config_path",
+        autospec=True,
+    )  # noqa: E501
+    @mock.patch(
+        "google.auth.transport._mtls_helper._read_cert_and_key_files",
+        autospec=True,
+    )  # noqa: E501
+    @mock.patch(
+        "google.auth.transport._mtls_helper.path.exists", autospec=True
+    )  # noqa: E501
+    def test_no_workload_fallback_to_home(
+        self,
+        mock_path_exists,
+        mock_read_cert_and_key_files,
+        mock_get_cert_config_path,
+        mock_load_json_file,
+    ):
+        ecp_path = "/etc/gcloud/certificate_config.json"
+        home_path = os.path.join(
+            _mtls_helper._cloud_sdk.get_config_path(),
+            "certificate_config.json",
+        )
+        mock_get_cert_config_path.return_value = ecp_path
+
+        def exists_side_effect(path):
+            if path == home_path:
+                return True
+            return False
+
+        mock_path_exists.side_effect = exists_side_effect
+
+        def load_json_side_effect(path):
+            if path == ecp_path:
+                return {"cert_configs": {"pkcs11": {}}}
+            elif path == home_path:
+                return {
+                    "cert_configs": {
+                        "workload": {
+                            "cert_path": "cert/path",
+                            "key_path": "key/path",
+                        }
+                    }
+                }
+            return {}
+
+        mock_load_json_file.side_effect = load_json_side_effect
+        mock_read_cert_and_key_files.return_value = (
+            pytest.public_cert_bytes,
+            pytest.private_key_bytes,
+        )
+
+        actual_cert, actual_key = _mtls_helper._get_workload_cert_and_key(None)
+        assert actual_cert == pytest.public_cert_bytes
+        assert actual_key == pytest.private_key_bytes
+
+        mock_get_cert_config_path.assert_called_once_with(None, True)
+        mock_load_json_file.assert_has_calls(
+            [mock.call(ecp_path), mock.call(home_path)]
+        )
+        mock_read_cert_and_key_files.assert_called_once_with(
+            "cert/path", "key/path"
+        )  # noqa: E501
+
+    @mock.patch(
+        "google.auth.transport._mtls_helper._load_json_file", autospec=True
+    )  # noqa: E501
+    @mock.patch(
+        "google.auth.transport._mtls_helper._get_cert_config_path",
+        autospec=True,
+    )  # noqa: E501
+    @mock.patch(
+        "google.auth.transport._mtls_helper._read_cert_and_key_files",
+        autospec=True,
+    )  # noqa: E501
+    @mock.patch(
+        "google.auth.transport._mtls_helper.path.exists", autospec=True
+    )  # noqa: E501
+    def test_no_workload_fallback_to_home_error(
+        self,
+        mock_path_exists,
+        mock_read_cert_and_key_files,
+        mock_get_cert_config_path,
+        mock_load_json_file,
+    ):
+        ecp_path = "/etc/gcloud/certificate_config.json"
+        home_path = os.path.join(
+            _mtls_helper._cloud_sdk.get_config_path(),
+            "certificate_config.json",
+        )
+        mock_get_cert_config_path.return_value = ecp_path
+
+        def exists_side_effect(path):
+            if path == home_path:
+                return True
+            return False
+
+        mock_path_exists.side_effect = exists_side_effect
+
+        def load_json_side_effect(path):
+            if path == ecp_path:
+                return {"cert_configs": {"pkcs11": {}}}
+            elif path == home_path:
+                raise exceptions.ClientCertError("mocked unreadable file")
+            return {}
+
+        mock_load_json_file.side_effect = load_json_side_effect
+
+        actual_cert, actual_key = _mtls_helper._get_workload_cert_and_key(None)
+        assert actual_cert is None
+        assert actual_key is None
+
+        mock_get_cert_config_path.assert_called_once_with(None, True)
+        mock_load_json_file.assert_has_calls(
+            [mock.call(ecp_path), mock.call(home_path)]
+        )
+        mock_read_cert_and_key_files.assert_not_called()
+
+    @mock.patch(
+        "google.auth.transport._mtls_helper._load_json_file", autospec=True
+    )  # noqa: E501
+    @mock.patch(
+        "google.auth.transport._mtls_helper._get_cert_config_path",
+        autospec=True,
+    )
+    @mock.patch(
+        "google.auth.transport._mtls_helper.path.exists", autospec=True
+    )  # noqa: E501
+    @mock.patch("os.path.normpath", autospec=True)
+    def test_no_workload_fallback_avoided_same_path_normalization(
+        self,
+        mock_normpath,
+        mock_path_exists,
+        mock_get_cert_config_path,
+        mock_load_json_file,
+    ):
+        ecp_path = "C:/Users/User/.config/gcloud/certificate_config.json"
+        home_path = "C:\\Users\\User\\.config\\gcloud/certificate_config.json"
+        mock_get_cert_config_path.return_value = ecp_path
+
+        mock_path_exists.return_value = True
+
+        # When resolving, the first file has no workload.
+        mock_load_json_file.return_value = {"cert_configs": {"pkcs11": {}}}
+
+        win_path = "C:\\Users\\User\\.config\\gcloud\\certificate_config.json"
+
+        # Mock normpath to return the same string for both paths,
+        # simulating Windows path normalization.
+        def normpath_side_effect(path):
+            if path in [ecp_path, home_path]:
+                return win_path
+            return path
+
+        mock_normpath.side_effect = normpath_side_effect
+
+        # Mock get_config_path to construct a path with backslashes
+        with mock.patch(
+            "google.auth._cloud_sdk.get_config_path",
+            return_value="C:\\Users\\User\\.config\\gcloud",
+        ):
+            actual_cert, actual_key = _mtls_helper._get_workload_cert_and_key(
+                None
+            )  # noqa: E501
+
+        assert actual_cert is None
+        assert actual_key is None
+
+        # Check that it resolved ECP path but never attempted to load
+        # home_path (because it normalized to the same file).
+        mock_get_cert_config_path.assert_called_once_with(None, True)
+        mock_load_json_file.assert_called_once_with(ecp_path)
+
     @mock.patch("google.auth.transport._mtls_helper._load_json_file", autospec=True)
     @mock.patch(
         "google.auth.transport._mtls_helper._get_cert_config_path", autospec=True