fix: fail-fast on missing ECP config file to avoid 30s hang (#17377)

This PR resolves two issues in the `google-auth` package:

First, it adds a fast-fail check for ECP configuration. When the
`GOOGLE_API_CERTIFICATE_CONFIG` environment variable is set but the
configuration file is missing (common on corporate workstations or clean
sandbox test runners), the SDK was falling through to the well-known
SPIFFE path and waiting on a 30-second retry loop. We now check if the
config path is set but missing, and if we are not in a workload
environment (the well-known credentials directory is absent), we
immediately return `None` to fallback to unbound tokens. (Fixes
b/512912028)

Second, it fixes an incorrect mock in `test_mtls.py`.
`test_default_client_encrypted_cert_source` was mocking `open` in the
test namespace instead of the target module namespace. This caused the
test to write actual `cert_path` and `key_path` files to the local disk
during test runs. This is fixed by patching
`google.auth.transport.mtls.open`.

Unit tests have been added to verify the fast-fail behavior, and
existing retry tests have been updated to mock the workload directory.
All tests now pass without writing files to disk.

Author: nbayati

packages/google-auth/google/auth/_agent_identity_utils.py

@@ -89,6 +89,14 @@ def get_agent_identity_certificate_path():
     if not cert_config_path and not has_well_known_dir:
         return None
 
+    # If ECP config path is specified but does not exist, and we are on a workstation, fail-fast immediately.
+    if (
+        cert_config_path
+        and not has_well_known_dir
+        and not os.path.exists(cert_config_path)
+    ):
+        return None
+
     has_logged_config_warning = False
     has_logged_cert_warning = False
 

packages/google-auth/tests/test_agent_identity_utils.py

@@ -165,29 +165,47 @@ def test_get_agent_identity_certificate_path_success(self, tmpdir, monkeypatch):
         assert result == str(cert_path)
 
     @mock.patch("time.sleep")
+    @mock.patch("google.auth._agent_identity_utils.os.path.exists")
     def test_get_agent_identity_certificate_path_retry(
-        self, mock_sleep, tmpdir, monkeypatch
+        self, mock_exists, mock_sleep, tmpdir, monkeypatch
     ):
         config_path = tmpdir.join("config.json")
         monkeypatch.setenv(
             environment_vars.GOOGLE_API_CERTIFICATE_CONFIG, str(config_path)
         )
 
+        # Simulate workload env (well_known_dir exists) to avoid fail-fast
+        def exists_side_effect(path):
+            if path == "/var/run/secrets/workload-spiffe-credentials":
+                return True
+            return False
+
+        mock_exists.side_effect = exists_side_effect
+
         # File doesn't exist initially
         with pytest.raises(exceptions.RefreshError):
             _agent_identity_utils.get_agent_identity_certificate_path()
 
         assert mock_sleep.call_count == len(_agent_identity_utils._POLLING_INTERVALS)
 
     @mock.patch("time.sleep")
+    @mock.patch("google.auth._agent_identity_utils.os.path.exists")
     def test_get_agent_identity_certificate_path_failure(
-        self, mock_sleep, tmpdir, monkeypatch
+        self, mock_exists, mock_sleep, tmpdir, monkeypatch
     ):
         config_path = tmpdir.join("non_existent_config.json")
         monkeypatch.setenv(
             environment_vars.GOOGLE_API_CERTIFICATE_CONFIG, str(config_path)
         )
 
+        # Simulate workload env (well_known_dir exists) to avoid fail-fast
+        def exists_side_effect(path):
+            if path == "/var/run/secrets/workload-spiffe-credentials":
+                return True
+            return False
+
+        mock_exists.side_effect = exists_side_effect
+
         with pytest.raises(exceptions.RefreshError) as excinfo:
             _agent_identity_utils.get_agent_identity_certificate_path()
 
@@ -198,6 +216,19 @@ def test_get_agent_identity_certificate_path_failure(
         )
         assert mock_sleep.call_count == len(_agent_identity_utils._POLLING_INTERVALS)
 
+    def test_get_agent_identity_certificate_path_workstation_fail_fast(
+        self, tmpdir, monkeypatch
+    ):
+        config_path = tmpdir.join("non_existent_config.json")
+        monkeypatch.setenv(
+            environment_vars.GOOGLE_API_CERTIFICATE_CONFIG, str(config_path)
+        )
+
+        # On a workstation, well_known_dir does not exist, and config file is missing.
+        # It should fail-fast and return None immediately.
+        result = _agent_identity_utils.get_agent_identity_certificate_path()
+        assert result is None
+
     @mock.patch("time.sleep")
     @mock.patch("os.path.exists")
     def test_get_agent_identity_certificate_path_cert_not_found(

packages/google-auth/tests/transport/test_mtls.py

@@ -154,8 +154,10 @@ def test_default_client_encrypted_cert_source(
     # Test good callback.
     get_client_ssl_credentials.return_value = (True, b"cert", b"key", b"passphrase")
     callback = mtls.default_client_encrypted_cert_source("cert_path", "key_path")
-    with mock.patch("{}.open".format(__name__), return_value=mock.MagicMock()):
+    with mock.patch("google.auth.transport.mtls.open", mock.mock_open()) as mock_file:
         assert callback() == ("cert_path", "key_path", b"passphrase")
+        mock_file.assert_any_call("cert_path", "wb")
+        mock_file.assert_any_call("key_path", "wb")
 
     # Test bad callback which throws exception.
     get_client_ssl_credentials.side_effect = exceptions.ClientCertError()