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

[attharva-24]

….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:

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)

[gemini-code-assist]

Code Review

This pull request introduces a fallback mechanism to load the workload certificate configuration from the default home path when it is not found in the initial configuration. It also adds error handling to raise a ClientCertError when the workload certificate configuration is missing, along with corresponding unit tests. The feedback suggests adding defensive type checks when loading the JSON configuration file to prevent potential TypeErrors if the file is malformed or empty.

[gemini-code-assist]

If the JSON file is empty, contains null, or is otherwise malformed, _load_json_file can return None or a non-dictionary object. This would cause a TypeError when checking "cert_configs" in home_data. We should defensively verify that home_data and home_data["cert_configs"] are dictionaries before accessing them.

Suggested change

	home_data = _load_json_file(default_home_path)
	if "cert_configs" in home_data and "workload" in home_data["cert_configs"]:
	cert_configs = home_data["cert_configs"]
	absolute_path = default_home_path
	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

[macastelaz]

I think that this causes problems, even with path.expanduser, despite the comment above it. I don't actually see this used anywhere else anyways, I'd suggest we get rid of it (can be done in a separate PR likely for better separation of concerns) and then here we can use

default_home_path = os.path.join(                                        
        _cloud_sdk.get_config_path(), "certificate_config.json"              
    )

This already ensures we handle environments with different filepath systems (e.g. windows) and when there are custom config directories setup (e.g. CLOUDSDK_CONFIG / CLOUD_SDK_CONFIG_DIR)

[macastelaz]

We should apply the same isInstance(workload, dict) check here to ensure the workload is a dictionary type before we eventually try to get values from the dict below (cert_parth and key_path).

[macastelaz]

This could throw and if it does, I think we still want to fallback so that get_client_ssl_credentials can move forward with other attempts. Wrapping this in a try ... except block to pass on exceptions so that we can return None, None still is likely desirable. E.g.

if path.exists(default_home_path) and default_home_path != 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

[macastelaz]

I think we also want to ensure cert_configs is a dict too and if it isn't, throw the invalid format exception to avoid something like {"cert_configs": "not_a_dict"} being allowed

[macastelaz]

nit: I think you can remove this and then on line 516 call _mtls_helper._get_workload_cert_and_key(None)

[macastelaz]

I think this test misses assertions that could help us prove the fallback worked the way we expect - something like:

            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")

[macastelaz]

I think now that this can raise a ClientCertError we may want to wrap this in a try...Except to re-raise this as a RefreshError which adheres to the contract of refresh (

google-cloud-python/packages/google-auth/google/auth/credentials.py

Line 167 in 2059bdf

google.auth.exceptions.RefreshError: If the credentials could

)