Skip to content

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

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
attharva-24 wants to merge 2 commits into
base: main
Choose a base branch
from
Open

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

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
5d527fb
fix(auth): prevent TypeError and support home-dir cert fallback for X…
attharva-24 Aug 6, 2026
8e68cf0
fix(auth): add defensive dictionary checks for loaded certificate con…
attharva-24 Aug 6, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
4 changes: 4 additions & 0 deletions packages/google-auth/google/auth/identity_pool.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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):
Expand Down
14 changes: 12 additions & 2 deletions packages/google-auth/google/auth/transport/_mtls_helper.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -459,7 +459,7 @@ 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:

@macastelaz macastelaz Aug 7, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

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

raise exceptions.ClientCertError(
'Certificate config file {} is in an invalid format, a "cert configs" object is expected'.format(
absolute_path
Expand All @@ -472,7 +472,17 @@ 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(CERTIFICATE_CONFIGURATION_DEFAULT_PATH)

@macastelaz macastelaz Aug 7, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

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)

if path.exists(default_home_path) and default_home_path != absolute_path:
home_data = _load_json_file(default_home_path)

@macastelaz macastelaz Aug 7, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

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

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

if not isinstance(cert_configs, dict) or "workload" not in cert_configs:
return None, None
workload = cert_configs["workload"]

Expand Down
16 changes: 16 additions & 0 deletions packages/google-auth/tests/test_identity_pool.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1784,6 +1784,22 @@ 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("google.auth._agent_identity_utils.parse_certificate")
@mock.patch(
"google.auth._agent_identity_utils.should_request_bound_token",
Expand Down
63 changes: 63 additions & 0 deletions packages/google-auth/tests/transport/test__mtls_helper.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -499,6 +499,22 @@ 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._get_cert_config_path", autospec=True
)
@mock.patch("os.path.exists", autospec=True)

@macastelaz macastelaz Aug 7, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

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

def test_malformed_json_returns_error(
self, mock_path_exists, mock_get_cert_config_path, mock_load_json_file
):
mock_path_exists.return_value = True
mock_get_cert_config_path.return_value = "/path/to/cert"

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

@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
Expand All @@ -511,6 +527,53 @@ 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)
@mock.patch(
"google.auth.transport._mtls_helper._get_cert_config_path", autospec=True
)
@mock.patch(
"google.auth.transport._mtls_helper._read_cert_and_key_files", autospec=True
)
@mock.patch("os.path.exists", autospec=True)
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.expanduser("~/.config/gcloud/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

@macastelaz macastelaz Aug 7, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

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


@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
Expand Down