feat(auth): support ECP hardware-backed certificates for X.509 WIF

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

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"

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

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)

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/<python-version> auth/<library-version> 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,

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(