diff --git a/packages/google-auth/google/auth/external_account.py b/packages/google-auth/google/auth/external_account.py index b90fcab4c0ee..74d7aedee2a2 100644 --- a/packages/google-auth/google/auth/external_account.py +++ b/packages/google-auth/google/auth/external_account.py @@ -479,9 +479,13 @@ def _perform_refresh_token(self, request, cert_fingerprint=None): # Inject client certificate into request. if self._mtls_required(): - request = functools.partial( - request, cert=self._get_mtls_cert_and_key_paths() - ) + cert_path, key_path = self._get_mtls_cert_and_key_paths() + # Only inject file-based cert/key when a key path is available. + # When using hardware-backed keys (ECP), the key is in the + # hardware keystore (key_path is None) and mTLS is handled by + # the session's ECP adapter instead. + if key_path is not None: + request = functools.partial(request, cert=(cert_path, key_path)) if self._should_initialize_impersonated_credentials(): with self._impersonation_lock: @@ -536,7 +540,8 @@ def _perform_refresh_token(self, request, cert_fingerprint=None): self.expiry = now + lifetime def _build_regional_access_boundary_lookup_url( - self, request: "Optional[google.auth.transport.Request]" = None # noqa: F821 + self, + request: "Optional[google.auth.transport.Request]" = None, # noqa: F821 ): """Builds and returns the URL for the Regional Access Boundary lookup API.""" if getattr(self, "_impersonated_credentials", None): @@ -746,7 +751,7 @@ def from_info(cls, info, **kwargs): "universe_domain", credentials.DEFAULT_UNIVERSE_DOMAIN ), trust_boundary=info.get("trust_boundary"), - **kwargs + **kwargs, ) @classmethod diff --git a/packages/google-auth/google/auth/identity_pool.py b/packages/google-auth/google/auth/identity_pool.py index dd5f103b7250..d681a8d11be4 100644 --- a/packages/google-auth/google/auth/identity_pool.py +++ b/packages/google-auth/google/auth/identity_pool.py @@ -277,7 +277,7 @@ def __init__( credential_source=None, subject_token_supplier=None, *args, - **kwargs + **kwargs, ): """Instantiates an external account credentials object from a file/URL. @@ -335,7 +335,7 @@ def __init__( token_url=token_url, credential_source=credential_source, *args, - **kwargs + **kwargs, ) if credential_source is None and subject_token_supplier is None: raise exceptions.InvalidValue( @@ -412,11 +412,20 @@ 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." + if cert_path is not None: + return _mtls_helper._read_cert_file(cert_path) + + if _mtls_helper.is_ecp_config(self._certificate_config_location): + from google.auth.transport import _custom_tls_signer + + config_path = _mtls_helper._get_cert_config_path( + self._certificate_config_location ) - return _mtls_helper._read_cert_file(cert_path) + return _custom_tls_signer.get_cert_from_custom_tls_signer(config_path) + + raise exceptions.ClientCertError( + "Workload certificate configuration could not be found or does not contain workload certificate paths." + ) def _mtls_required(self): return self._credential_source_certificate is not None @@ -574,7 +583,11 @@ def refresh(self, request): if self._credential_source_certificate is not None: try: cert_bytes = self._get_cert_bytes() - except (exceptions.ClientCertError, OSError) as e: + except ( + exceptions.ClientCertError, + exceptions.MutualTLSChannelError, + OSError, + ) as e: raise exceptions.RefreshError( "Failed to retrieve certificate bytes for external" " account credentials" diff --git a/packages/google-auth/google/auth/transport/_custom_tls_signer.py b/packages/google-auth/google/auth/transport/_custom_tls_signer.py index 90143101ab07..b6e1ea118210 100644 --- a/packages/google-auth/google/auth/transport/_custom_tls_signer.py +++ b/packages/google-auth/google/auth/transport/_custom_tls_signer.py @@ -207,6 +207,52 @@ def get_cert(signer_lib, config_file_path): return bytes(cert_holder) +def get_cert_from_custom_tls_signer(enterprise_cert_file_path): + """Retrieves certificate bytes from the signer library specified in the enterprise cert JSON file. + + Args: + enterprise_cert_file_path (str): The path to an enterprise cert JSON file. + + Returns: + bytes: The certificate bytes in PEM format. + + Raises: + google.auth.exceptions.MutualTLSChannelError: If signer library is missing or fails. + """ + try: + with open(enterprise_cert_file_path, "r", encoding="utf-8") as f: + enterprise_cert_json = json.load(f) + except (FileNotFoundError, OSError, json.JSONDecodeError) as e: + raise exceptions.MutualTLSChannelError( + f"Failed to read enterprise cert file at {enterprise_cert_file_path}" + ) from e + + if not isinstance(enterprise_cert_json, dict): + raise exceptions.MutualTLSChannelError("enterprise cert file is invalid") + + libs = enterprise_cert_json.get("libs") + if not isinstance(libs, dict): + raise exceptions.MutualTLSChannelError( + "enterprise cert file is missing 'libs' section" + ) + + signer_library = ( + libs.get("ecp_client") or libs.get("signer_library") or libs.get("ecp") + ) + if not isinstance(signer_library, str) or not signer_library: + raise exceptions.MutualTLSChannelError( + "enterprise cert file is missing signer library (ecp_client)" + ) + + try: + signer_lib = load_signer_lib(signer_library) + return get_cert(signer_lib, enterprise_cert_file_path) + except (OSError, AttributeError, TypeError, ValueError) as e: + raise exceptions.MutualTLSChannelError( + f"Failed to load or execute signer library at {signer_library}: {e}" + ) from e + + class CustomTlsSigner(object): def __init__(self, enterprise_cert_file_path): """ diff --git a/packages/google-auth/google/auth/transport/_mtls_helper.py b/packages/google-auth/google/auth/transport/_mtls_helper.py index 7779c484c713..600bfc292c08 100644 --- a/packages/google-auth/google/auth/transport/_mtls_helper.py +++ b/packages/google-auth/google/auth/transport/_mtls_helper.py @@ -35,6 +35,7 @@ # Default gcloud config path, to be used with path.expanduser for cross-platform compatibility. CERTIFICATE_CONFIGURATION_DEFAULT_PATH = "~/.config/gcloud/certificate_config.json" _CERT_PROVIDER_COMMAND = "cert_provider_command" +_ECP_SECTIONS = ("pkcs11", "windows_store", "macos_keychain") _CERT_REGEX = re.compile( b"-----BEGIN CERTIFICATE-----.+-----END CERTIFICATE-----\r?\n?", re.DOTALL ) @@ -132,9 +133,11 @@ def secure_cert_key_paths( key_path is None or os.path.exists(key_path) ): if _can_read(cert_path) and _can_read(key_path): - yield cast(str, cert_path or cert), cast( - str, key_path or key - ), passphrase + yield ( + cast(str, cert_path or cert), + cast(str, key_path or key), + passphrase, + ) return except _MemfdCreationError: pass # Fallback to Tier 3 on failure. @@ -521,6 +524,43 @@ def _get_workload_cert_and_key_paths(config_path, include_context_aware=True): return cert_path, key_path +def is_ecp_config(certificate_config_path=None, include_context_aware=True): + """Checks if the given configuration is an ECP (Enterprise Certificate Proxy) certificate configuration. + + Args: + certificate_config_path (Optional[str]): The certificate config path. If no path is provided, + the environment variable will be checked first, then the well known gcloud location. + include_context_aware (bool): If context aware metadata path should be checked for the + SecureConnect mTLS configuration. + + Returns: + bool: True if configuration specifies ECP, False otherwise. + """ + config_path = _get_cert_config_path(certificate_config_path, include_context_aware) + if config_path is None: + return False + + try: + data = _load_json_file(config_path) + except (exceptions.ClientCertError, OSError): + return False + + if not isinstance(data, dict): + return False + + cert_configs = data.get("cert_configs") + if not isinstance(cert_configs, dict): + return False + + if "workload" in cert_configs: + return False + + has_ecp_section = any(section in cert_configs for section in _ECP_SECTIONS) + has_libs = isinstance(data.get("libs"), dict) + + return has_ecp_section and has_libs + + def _read_cert_and_key_files(cert_path, key_path): cert_data = _read_cert_file(cert_path) key_data = _read_key_file(key_path) diff --git a/packages/google-auth/tests/test_external_account.py b/packages/google-auth/tests/test_external_account.py index a637a95cf168..b0d222dfadee 100644 --- a/packages/google-auth/tests/test_external_account.py +++ b/packages/google-auth/tests/test_external_account.py @@ -779,6 +779,55 @@ def test_refresh_with_mtls( assert not credentials.expired assert credentials.token == response["access_token"] + @mock.patch( + "google.auth.metrics.python_and_auth_lib_version", + return_value=LANG_LIBRARY_METRICS_HEADER_VALUE, + ) + @mock.patch("google.auth._helpers.utcnow", return_value=datetime.datetime.min) + @mock.patch( + "google.auth.external_account.Credentials._mtls_required", return_value=True + ) + @mock.patch( + "google.auth.external_account.Credentials._get_mtls_cert_and_key_paths", + return_value=(None, None), + ) + def test_refresh_with_mtls_ecp_no_key( + self, + mock_get_mtls_cert_and_key_paths, + mock_mtls_required, + unused_utcnow, + mock_auth_lib_value, + ): + response = self.SUCCESS_RESPONSE.copy() + response["expires_in"] = 2800 + expected_expiry = datetime.datetime.min + datetime.timedelta( + seconds=response["expires_in"] + ) + headers = { + "Content-Type": "application/x-www-form-urlencoded", + "x-goog-api-client": "gl-python/ auth/ google-byoid-sdk sa-impersonation/false config-lifetime/false", + } + request_data = { + "grant_type": "urn:ietf:params:oauth:grant-type:token-exchange", + "audience": self.AUDIENCE, + "requested_token_type": "urn:ietf:params:oauth:token-type:access_token", + "subject_token": "subject_token_0", + "subject_token_type": self.SUBJECT_TOKEN_TYPE, + } + request = self.make_mock_request(status=http_client.OK, data=response) + credentials = self.make_credentials() + + credentials.refresh(request) + + # Expected cert path is None because ECP uses transport adapter for mTLS + self.assert_token_request_kwargs( + request.call_args[1], headers, request_data, None + ) + assert credentials.valid + assert credentials.expiry == expected_expiry + assert not credentials.expired + assert credentials.token == response["access_token"] + @mock.patch( "google.auth.metrics.python_and_auth_lib_version", return_value=LANG_LIBRARY_METRICS_HEADER_VALUE, diff --git a/packages/google-auth/tests/test_identity_pool.py b/packages/google-auth/tests/test_identity_pool.py index 1138db284db7..6828fc259db5 100644 --- a/packages/google-auth/tests/test_identity_pool.py +++ b/packages/google-auth/tests/test_identity_pool.py @@ -1784,22 +1784,72 @@ 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.is_ecp_config", + return_value=False, + ) @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 + def test_retrieve_subject_token_none_raises_error( + self, mock_get_workload_cert_and_key_paths, mock_is_ecp_config + ): + credentials = self.make_credentials( + credential_source=self.CREDENTIAL_SOURCE_CERTIFICATE.copy() + ) + + with pytest.raises(exceptions.RefreshError) as excinfo: + credentials.retrieve_subject_token(None) + + assert excinfo.match("Failed to retrieve leaf certificate.") + + @mock.patch( + "google.auth.transport._custom_tls_signer.get_cert_from_custom_tls_signer", + return_value=open(CERT_FILE, "rb").read(), + ) + @mock.patch( + "google.auth.transport._mtls_helper._get_cert_config_path", + return_value="/path/to/ecp_config.json", + ) + @mock.patch("google.auth.transport._mtls_helper.is_ecp_config", return_value=True) + @mock.patch( + "google.auth.transport._mtls_helper._get_workload_cert_and_key_paths", + return_value=(None, None), + ) + def test_retrieve_subject_token_ecp_success( + self, + mock_get_workload_paths, + mock_is_ecp_config, + mock_get_cert_config_path, + mock_get_cert_from_ecp, + ): + credentials = self.make_credentials( + credential_source=self.CREDENTIAL_SOURCE_CERTIFICATE.copy() + ) + subject_token = credentials.retrieve_subject_token(None) + assert subject_token == json.dumps([CERT_FILE_CONTENT]) + mock_is_ecp_config.assert_called_once() + mock_get_cert_config_path.assert_called_once() + mock_get_cert_from_ecp.assert_called_once_with("/path/to/ecp_config.json") + + @mock.patch.object( + identity_pool.Credentials, + "_get_cert_bytes", + side_effect=exceptions.MutualTLSChannelError("mock mtls error"), + ) + def test_refresh_mutual_tls_channel_error_raises_refresh_error( + self, mock_get_cert_bytes ): credentials = self.make_credentials( credential_source=self.CREDENTIAL_SOURCE_CERTIFICATE.copy() ) - with pytest.raises(exceptions.ClientCertError) as excinfo: - credentials._get_cert_bytes() + with pytest.raises(exceptions.RefreshError) as excinfo: + credentials.refresh(None) assert excinfo.match( - "Workload certificate configuration could not be found or does not contain workload certificate paths." + "Failed to retrieve certificate bytes for external account credentials" ) @mock.patch.object( diff --git a/packages/google-auth/tests/transport/test__custom_tls_signer.py b/packages/google-auth/tests/transport/test__custom_tls_signer.py index fa210ee0b8d7..fba1743ce5ce 100644 --- a/packages/google-auth/tests/transport/test__custom_tls_signer.py +++ b/packages/google-auth/tests/transport/test__custom_tls_signer.py @@ -299,8 +299,9 @@ def test_cast_ssl_ctx_to_void_p_stdlib_unsupported_runtime_trace_refs(): fake_impl = mock.Mock() fake_impl.name = "cpython" - with mock.patch("sys.implementation", fake_impl), mock.patch( - "sys.getobjects", create=True + with ( + mock.patch("sys.implementation", fake_impl), + mock.patch("sys.getobjects", create=True), ): with pytest.raises( exceptions.MutualTLSChannelError, @@ -320,8 +321,9 @@ def test_cast_ssl_ctx_to_void_p_stdlib_unsupported_runtime_debug_flag(): context = ssl.SSLContext() fake_impl = mock.Mock() fake_impl.name = "cpython" - with mock.patch("sys.implementation", fake_impl), mock.patch( - "sysconfig.get_config_var", return_value=1 + with ( + mock.patch("sys.implementation", fake_impl), + mock.patch("sysconfig.get_config_var", return_value=1), ): with pytest.raises( exceptions.MutualTLSChannelError, @@ -340,8 +342,9 @@ def mock_get_config_var(var): fake_impl = mock.Mock() fake_impl.name = "cpython" - with mock.patch("sys.implementation", fake_impl), mock.patch( - "sysconfig.get_config_var", side_effect=mock_get_config_var + with ( + mock.patch("sys.implementation", fake_impl), + mock.patch("sysconfig.get_config_var", side_effect=mock_get_config_var), ): with pytest.raises( exceptions.MutualTLSChannelError, @@ -366,3 +369,66 @@ def test_cast_ssl_ctx_to_void_p_stdlib_mock_error(): TypeError, match="context must be an instance of ssl.SSLContext, not a mock" ): _custom_tls_signer._cast_ssl_ctx_to_void_p_stdlib(context) + + +def test_get_cert_from_custom_tls_signer_success(): + signer_lib = mock.MagicMock() + with mock.patch( + "google.auth.transport._custom_tls_signer.load_signer_lib", + return_value=signer_lib, + ) as mock_load_lib: + with mock.patch( + "google.auth.transport._custom_tls_signer.get_cert", + return_value=b"mock_cert_bytes", + ) as mock_get_cert: + cert = _custom_tls_signer.get_cert_from_custom_tls_signer( + ENTERPRISE_CERT_FILE + ) + assert cert == b"mock_cert_bytes" + mock_load_lib.assert_called_once_with("/path/to/signer/lib") + mock_get_cert.assert_called_once_with(signer_lib, ENTERPRISE_CERT_FILE) + + +def test_get_cert_from_custom_tls_signer_missing_libs(tmp_path): + config_file = tmp_path / "cert_config.json" + config_file.write_text('{"cert_configs": {}}') + with pytest.raises( + exceptions.MutualTLSChannelError, match="missing 'libs' section" + ): + _custom_tls_signer.get_cert_from_custom_tls_signer(str(config_file)) + + +def test_get_cert_from_custom_tls_signer_missing_signer_lib(): + with pytest.raises( + exceptions.MutualTLSChannelError, match="missing signer library" + ): + _custom_tls_signer.get_cert_from_custom_tls_signer(INVALID_ENTERPRISE_CERT_FILE) + + +def test_get_cert_from_custom_tls_signer_load_lib_oserror(): + with mock.patch( + "google.auth.transport._custom_tls_signer.load_signer_lib", + side_effect=OSError("cannot open shared object file"), + ): + with pytest.raises( + exceptions.MutualTLSChannelError, + match="Failed to load or execute signer library", + ): + _custom_tls_signer.get_cert_from_custom_tls_signer(ENTERPRISE_CERT_FILE) + + +def test_get_cert_from_custom_tls_signer_get_cert_attribute_error(): + signer_lib = mock.MagicMock() + with mock.patch( + "google.auth.transport._custom_tls_signer.load_signer_lib", + return_value=signer_lib, + ): + with mock.patch( + "google.auth.transport._custom_tls_signer.get_cert", + side_effect=AttributeError("function not found in library"), + ): + with pytest.raises( + exceptions.MutualTLSChannelError, + match="Failed to load or execute signer library", + ): + _custom_tls_signer.get_cert_from_custom_tls_signer(ENTERPRISE_CERT_FILE) diff --git a/packages/google-auth/tests/transport/test__mtls_helper.py b/packages/google-auth/tests/transport/test__mtls_helper.py index e9bb62db2133..5fc50d3829bb 100644 --- a/packages/google-auth/tests/transport/test__mtls_helper.py +++ b/packages/google-auth/tests/transport/test__mtls_helper.py @@ -1302,8 +1302,9 @@ def test_memfd_success(self, mock_memfd_cm, mock_memfd_create): ) mock_memfd_cm.return_value = mock_memfd_ctx - with mock.patch.object(os.path, "exists", return_value=True), mock.patch( - "builtins.open", mock.mock_open() + with ( + mock.patch.object(os.path, "exists", return_value=True), + mock.patch("builtins.open", mock.mock_open()), ): with _mtls_helper.secure_cert_key_paths( pytest.public_cert_bytes, @@ -1368,9 +1369,10 @@ def test_falls_back_to_tempfile_when_filesystem_unreadable( ) mock_tempfile_cm.return_value = mock_tempfile_ctx - with mock.patch.object(os.path, "exists", return_value=True), mock.patch( - "builtins.open", mock.mock_open() - ) as mock_open: + with ( + mock.patch.object(os.path, "exists", return_value=True), + mock.patch("builtins.open", mock.mock_open()) as mock_open, + ): mock_open.side_effect = PermissionError("Permission denied") with _mtls_helper.secure_cert_key_paths( @@ -1888,3 +1890,85 @@ def test_remove_oserror_ignored( mock_fh.flush.assert_called_once() mock_fsync.assert_called_once() mock_remove.assert_called_once_with("/path/to/secret") + + +class TestIsECPConfig: + @mock.patch("google.auth.transport._mtls_helper._load_json_file") + @mock.patch("google.auth.transport._mtls_helper._get_cert_config_path") + def test_is_ecp_config_pkcs11(self, mock_get_path, mock_load_json): + mock_get_path.return_value = "/path/to/config.json" + mock_load_json.return_value = { + "cert_configs": {"pkcs11": {"module": "/path/to/mod.so"}}, + "libs": {"ecp_client": "/path/to/lib.so"}, + } + assert _mtls_helper.is_ecp_config("/path/to/config.json") is True + + @mock.patch("google.auth.transport._mtls_helper._load_json_file") + @mock.patch("google.auth.transport._mtls_helper._get_cert_config_path") + def test_is_ecp_config_macos_keychain(self, mock_get_path, mock_load_json): + mock_get_path.return_value = "/path/to/config.json" + mock_load_json.return_value = { + "cert_configs": {"macos_keychain": {"issuer": "Corp CA"}}, + "libs": {"ecp_client": "/path/to/lib.dylib"}, + } + assert _mtls_helper.is_ecp_config("/path/to/config.json") is True + + @mock.patch("google.auth.transport._mtls_helper._load_json_file") + @mock.patch("google.auth.transport._mtls_helper._get_cert_config_path") + def test_is_ecp_config_windows_store(self, mock_get_path, mock_load_json): + mock_get_path.return_value = "/path/to/config.json" + mock_load_json.return_value = { + "cert_configs": {"windows_store": {"store": "MY"}}, + "libs": {"ecp_client": "C:\\path\\lib.dll"}, + } + assert _mtls_helper.is_ecp_config("/path/to/config.json") is True + + @mock.patch("google.auth.transport._mtls_helper._load_json_file") + @mock.patch("google.auth.transport._mtls_helper._get_cert_config_path") + def test_is_ecp_config_workload_returns_false(self, mock_get_path, mock_load_json): + mock_get_path.return_value = "/path/to/config.json" + mock_load_json.return_value = { + "cert_configs": { + "workload": {"cert_path": "cert.pem", "key_path": "key.pem"}, + "pkcs11": {"module": "/path/to/mod.so"}, + }, + "libs": {"ecp_client": "/path/to/lib.so"}, + } + assert _mtls_helper.is_ecp_config("/path/to/config.json") is False + + @mock.patch("google.auth.transport._mtls_helper._load_json_file") + @mock.patch("google.auth.transport._mtls_helper._get_cert_config_path") + def test_is_ecp_config_missing_libs_returns_false( + self, mock_get_path, mock_load_json + ): + mock_get_path.return_value = "/path/to/config.json" + mock_load_json.return_value = { + "cert_configs": {"pkcs11": {"module": "/path/to/mod.so"}} + } + assert _mtls_helper.is_ecp_config("/path/to/config.json") is False + + @mock.patch("google.auth.transport._mtls_helper._load_json_file") + @mock.patch("google.auth.transport._mtls_helper._get_cert_config_path") + def test_is_ecp_config_no_ecp_section_returns_false( + self, mock_get_path, mock_load_json + ): + mock_get_path.return_value = "/path/to/config.json" + mock_load_json.return_value = { + "cert_configs": {"other_section": {}}, + "libs": {"ecp_client": "/path/to/lib.so"}, + } + assert _mtls_helper.is_ecp_config("/path/to/config.json") is False + + @mock.patch("google.auth.transport._mtls_helper._get_cert_config_path") + def test_is_ecp_config_none_path(self, mock_get_path): + mock_get_path.return_value = None + assert _mtls_helper.is_ecp_config(None) is False + + @mock.patch("google.auth.transport._mtls_helper._load_json_file") + @mock.patch("google.auth.transport._mtls_helper._get_cert_config_path") + def test_is_ecp_config_load_error_returns_false( + self, mock_get_path, mock_load_json + ): + mock_get_path.return_value = "/path/to/config.json" + mock_load_json.side_effect = exceptions.ClientCertError("error") + assert _mtls_helper.is_ecp_config("/path/to/config.json") is False