diff --git a/GEMINI.md b/GEMINI.md new file mode 100644 index 000000000000..b9f1edb586e9 --- /dev/null +++ b/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`. diff --git a/packages/google-auth/google/auth/identity_pool.py b/packages/google-auth/google/auth/identity_pool.py index 4b1aa393b2fa..dd5f103b7250 100644 --- a/packages/google-auth/google/auth/identity_pool.py +++ b/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 = ( diff --git a/packages/google-auth/google/auth/transport/_mtls_helper.py b/packages/google-auth/google/auth/transport/_mtls_helper.py index eb0600740c0d..7779c484c713 100644 --- a/packages/google-auth/google/auth/transport/_mtls_helper.py +++ b/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 diff --git a/packages/google-auth/tests/test_identity_pool.py b/packages/google-auth/tests/test_identity_pool.py index 92659bd90b38..1138db284db7 100644 --- a/packages/google-auth/tests/test_identity_pool.py +++ b/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", diff --git a/packages/google-auth/tests/transport/test__mtls_helper.py b/packages/google-auth/tests/transport/test__mtls_helper.py index 537ef47e7295..e9bb62db2133 100644 --- a/packages/google-auth/tests/transport/test__mtls_helper.py +++ b/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