fix(auth): fail-fast on invalid or non-workload certificate configs in agent identity discovery (#17116)

nbayati · web-flow · commit f27a546127cb · 2026-05-15T08:07:57.000-04:00
The `GOOGLE_API_CERTIFICATE_CONFIG` environment variable is shared between Managed Workload Identity (MWLID) token-binding and other flows like Enterprise Certificate Provider (ECP) configs (e.g., PKCS#11). When this env var is set, agent identity discovery is triggered. If the config file exists but lacks the `"workload"` section (as with ECP configurations),we should exit early and return `None` to avoid delaying non-workload flows. In addition, if the config file on disk had syntax errors or invalid JSON, the previous logic entered a 30-second blocking retry loop before failing with `RefreshError`. To resolve this, the lookup logic now assumes that if the config file exists on disk, it is in its final format. If the file exists but lacks a `"workload"` section, has syntax errors, or is unreadable, we return `None` immediately to fail-fast and avoid startup delays. b/512912028 fixes #17145
diff --git a/packages/google-auth/google/auth/_agent_identity_utils.py b/packages/google-auth/google/auth/_agent_identity_utils.py
@@ -80,41 +80,86 @@ def get_agent_identity_certificate_path():
     import json
 
     cert_config_path = os.environ.get(environment_vars.GOOGLE_API_CERTIFICATE_CONFIG)
-    if not cert_config_path:
+
+    # Check if the well-known workload directory is mounted.
+    well_known_dir = os.path.dirname(_WELL_KNOWN_CERT_PATH)
+    has_well_known_dir = os.path.exists(well_known_dir)
+
+    # If we have neither a config path nor a well-known mount directory, exit immediately.
+    if not cert_config_path and not has_well_known_dir:
         return None
 
-    has_logged_warning = False
+    has_logged_config_warning = False
+    has_logged_cert_warning = False
 
     for interval in _POLLING_INTERVALS:
         try:
-            with open(cert_config_path, "r") as f:
-                cert_config = json.load(f)
-                cert_path = (
-                    cert_config.get("cert_configs", {})
-                    .get("workload", {})
-                    .get("cert_path")
+            # Path A: Config file is explicitly set
+            if cert_config_path:
+                with open(cert_config_path, "r") as f:
+                    cert_config = json.load(f)
+
+                cert_configs = (
+                    cert_config.get("cert_configs")
+                    if isinstance(cert_config, dict)
+                    else None
                 )
+                workload_config = (
+                    cert_configs.get("workload")
+                    if isinstance(cert_configs, dict)
+                    else None
+                )
+
+                if (
+                    not isinstance(workload_config, dict)
+                    or "cert_path" not in workload_config
+                ):
+                    return None
+
+                cert_path = workload_config["cert_path"]
                 if _is_certificate_file_ready(cert_path):
                     return cert_path
-        except (IOError, ValueError, KeyError):
-            if not has_logged_warning:
+
+                # The config was parsed, but the cert file is not ready yet
+                target_path = cert_path
+
+            # Path B: Config is NOT set, fallback to the well-known path
+            else:
+                if _is_certificate_file_ready(_WELL_KNOWN_CERT_PATH):
+                    return _WELL_KNOWN_CERT_PATH
+
+                # The well-known cert file is not ready yet
+                target_path = _WELL_KNOWN_CERT_PATH
+
+            # Log a warning on the first failed attempt to load the certificate file
+            if not has_logged_cert_warning:
                 _LOGGER.warning(
-                    "Certificate config file not found at %s (from %s environment "
-                    "variable). Retrying for up to %s seconds.",
-                    cert_config_path,
+                    "Certificate file not ready at %s. Retrying until startup timeout (up to %s seconds total)...",
+                    target_path,
+                    _TOTAL_TIMEOUT,
+                )
+                has_logged_cert_warning = True
+
+        except (IOError, ValueError, KeyError) as e:
+            if cert_config_path and os.path.exists(cert_config_path):
+                # If the file exists but has invalid JSON or is unreadable,
+                # we assume it is in its final format and fail-fast by returning None.
+                return None
+
+            if not has_logged_config_warning and cert_config_path:
+                _LOGGER.warning(
+                    "Certificate config file not found or incomplete: %s (from %s "
+                    "environment variable). Retrying until startup timeout (up to %s seconds total)...",
+                    e,
                     environment_vars.GOOGLE_API_CERTIFICATE_CONFIG,
                     _TOTAL_TIMEOUT,
                 )
-                has_logged_warning = True
+                has_logged_config_warning = True
             pass
 
-        # As a fallback, check the well-known certificate path.
-        if _is_certificate_file_ready(_WELL_KNOWN_CERT_PATH):
-            return _WELL_KNOWN_CERT_PATH
-
         # A sleep is required in two cases:
         # 1. The config file is not found (the except block).
-        # 2. The config file is found, but the certificate is not yet available.
+        # 2. The config file/well-known path is found, but the certificate is not yet available.
         # In both cases, we need to poll, so we sleep on every iteration
         # that doesn't return a certificate.
         time.sleep(interval)
diff --git a/packages/google-auth/noxfile.py b/packages/google-auth/noxfile.py
@@ -196,6 +196,7 @@ def docs(session):
     """Build the docs for this library."""
 
     session.install("-e", ".[aiohttp]")
+    session.install("requests==2.31.0")
     session.install("sphinx", "alabaster", "recommonmark", "sphinx-docstring-typing")
 
     shutil.rmtree(os.path.join("docs", "_build"), ignore_errors=True)
diff --git a/packages/google-auth/tests/test_agent_identity_utils.py b/packages/google-auth/tests/test_agent_identity_utils.py
@@ -172,7 +172,7 @@ def test_get_agent_identity_certificate_path_retry(
         with pytest.raises(exceptions.RefreshError):
             _agent_identity_utils.get_agent_identity_certificate_path()
 
-        assert mock_sleep.call_count == 100
+        assert mock_sleep.call_count == len(_agent_identity_utils._POLLING_INTERVALS)
 
     @mock.patch("time.sleep")
     def test_get_agent_identity_certificate_path_failure(
@@ -191,7 +191,7 @@ def test_get_agent_identity_certificate_path_failure(
             environment_vars.GOOGLE_API_PREVENT_AGENT_TOKEN_SHARING_FOR_GCP_SERVICES
             in str(excinfo.value)
         )
-        assert mock_sleep.call_count == 100
+        assert mock_sleep.call_count == len(_agent_identity_utils._POLLING_INTERVALS)
 
     @mock.patch("time.sleep")
     @mock.patch("os.path.exists")
@@ -215,7 +215,149 @@ def exists_side_effect(path):
         with pytest.raises(exceptions.RefreshError):
             _agent_identity_utils.get_agent_identity_certificate_path()
 
-        assert mock_sleep.call_count == 100
+        assert mock_sleep.call_count == len(_agent_identity_utils._POLLING_INTERVALS)
+
+    @mock.patch("time.sleep")
+    def test_get_agent_identity_certificate_path_non_workload_config(
+        self, mock_sleep, tmpdir, monkeypatch
+    ):
+        config_path = tmpdir.join("config.json")
+        config_path.write(
+            json.dumps({"cert_configs": {"pkcs11": {"module": "some_module"}}})
+        )
+        monkeypatch.setenv(
+            environment_vars.GOOGLE_API_CERTIFICATE_CONFIG, str(config_path)
+        )
+
+        result = _agent_identity_utils.get_agent_identity_certificate_path()
+
+        # Should return None immediately without polling
+        assert result is None
+        mock_sleep.assert_not_called()
+
+    @mock.patch("time.sleep")
+    def test_get_agent_identity_certificate_path_invalid_json(
+        self, mock_sleep, tmpdir, monkeypatch
+    ):
+        config_path = tmpdir.join("config.json")
+        config_path.write("{invalid_json: true}")  # Invalid JSON missing quotes
+        monkeypatch.setenv(
+            environment_vars.GOOGLE_API_CERTIFICATE_CONFIG, str(config_path)
+        )
+
+        result = _agent_identity_utils.get_agent_identity_certificate_path()
+
+        # Should return None immediately without polling/sleeping
+        assert result is None
+        mock_sleep.assert_not_called()
+
+    @mock.patch("time.sleep")
+    def test_get_agent_identity_certificate_path_non_dict_json(
+        self, mock_sleep, tmpdir, monkeypatch
+    ):
+        config_path = tmpdir.join("config.json")
+        config_path.write(json.dumps(["not", "a", "dict"]))  # Valid JSON list
+        monkeypatch.setenv(
+            environment_vars.GOOGLE_API_CERTIFICATE_CONFIG, str(config_path)
+        )
+
+        result = _agent_identity_utils.get_agent_identity_certificate_path()
+
+        # Should fail fast immediately and return None without polling
+        assert result is None
+        mock_sleep.assert_not_called()
+
+    @mock.patch("time.sleep")
+    def test_get_agent_identity_certificate_path_workload_config_missing_cert_path(
+        self, mock_sleep, tmpdir, monkeypatch
+    ):
+        config_path = tmpdir.join("config.json")
+        config_path.write(json.dumps({"cert_configs": {"workload": {}}}))
+        monkeypatch.setenv(
+            environment_vars.GOOGLE_API_CERTIFICATE_CONFIG, str(config_path)
+        )
+
+        result = _agent_identity_utils.get_agent_identity_certificate_path()
+
+        # Should return None immediately without polling
+        assert result is None
+        mock_sleep.assert_not_called()
+
+    @mock.patch("time.sleep")
+    @mock.patch("os.path.exists")
+    @mock.patch("google.auth._agent_identity_utils._is_certificate_file_ready")
+    def test_get_agent_identity_certificate_path_no_config_but_has_well_known_dir(
+        self, mock_is_ready, mock_exists, mock_sleep, monkeypatch
+    ):
+        monkeypatch.delenv(
+            environment_vars.GOOGLE_API_CERTIFICATE_CONFIG, raising=False
+        )
+
+        # Simulate that the well-known workload mount directory exists, and the cert is ready
+        mock_exists.return_value = True
+        mock_is_ready.return_value = True
+
+        result = _agent_identity_utils.get_agent_identity_certificate_path()
+
+        # Should return the well-known path immediately
+        assert result == _agent_identity_utils._WELL_KNOWN_CERT_PATH
+        mock_sleep.assert_not_called()
+
+    @mock.patch("time.sleep")
+    @mock.patch("os.path.exists")
+    def test_get_agent_identity_certificate_path_no_config_no_well_known_dir(
+        self, mock_exists, mock_sleep, monkeypatch
+    ):
+        monkeypatch.delenv(
+            environment_vars.GOOGLE_API_CERTIFICATE_CONFIG, raising=False
+        )
+
+        # Simulate that the well-known mount directory does NOT exist
+        mock_exists.return_value = False
+
+        result = _agent_identity_utils.get_agent_identity_certificate_path()
+
+        # Should return None immediately without polling
+        assert result is None
+        mock_sleep.assert_not_called()
+
+    @mock.patch("time.sleep")
+    @mock.patch("os.path.exists")
+    @mock.patch("google.auth._agent_identity_utils._is_certificate_file_ready")
+    def test_get_agent_identity_certificate_path_no_config_well_known_polling_success(
+        self, mock_is_ready, mock_exists, mock_sleep, monkeypatch
+    ):
+        monkeypatch.delenv(
+            environment_vars.GOOGLE_API_CERTIFICATE_CONFIG, raising=False
+        )
+
+        # Simulate that the directory exists, file appears on 2nd try
+        mock_exists.return_value = True
+        mock_is_ready.side_effect = [False, True]
+
+        result = _agent_identity_utils.get_agent_identity_certificate_path()
+
+        assert result == _agent_identity_utils._WELL_KNOWN_CERT_PATH
+        assert mock_sleep.call_count == 1
+
+    @mock.patch("time.sleep")
+    @mock.patch("os.path.exists")
+    @mock.patch("google.auth._agent_identity_utils._is_certificate_file_ready")
+    def test_get_agent_identity_certificate_path_no_config_well_known_polling_timeout(
+        self, mock_is_ready, mock_exists, mock_sleep, monkeypatch
+    ):
+        monkeypatch.delenv(
+            environment_vars.GOOGLE_API_CERTIFICATE_CONFIG, raising=False
+        )
+
+        # Simulate that the directory exists, but file never appears
+        mock_exists.return_value = True
+        mock_is_ready.return_value = False
+
+        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("google.auth._agent_identity_utils.get_agent_identity_certificate_path")
     def test_get_and_parse_agent_identity_certificate_opted_out(
@@ -261,27 +403,6 @@ def test_get_and_parse_agent_identity_certificate_success(
         mock_parse_certificate.assert_called_once_with(b"cert_bytes")
         assert result == mock_parse_certificate.return_value
 
-    @mock.patch("time.sleep", return_value=None)
-    @mock.patch("google.auth._agent_identity_utils._is_certificate_file_ready")
-    def test_get_agent_identity_certificate_path_fallback_to_well_known_path(
-        self, mock_is_ready, mock_sleep, monkeypatch
-    ):
-        # Set a dummy config path that won't be found.
-        monkeypatch.setenv(
-            environment_vars.GOOGLE_API_CERTIFICATE_CONFIG, "/dummy/config.json"
-        )
-
-        # First, the primary path from the (mocked) config is not ready.
-        # Then, the fallback well-known path is ready.
-        mock_is_ready.side_effect = [False, True]
-
-        result = _agent_identity_utils.get_agent_identity_certificate_path()
-
-        assert result == _agent_identity_utils._WELL_KNOWN_CERT_PATH
-        # The sleep should have been called once before the fallback is checked.
-        mock_sleep.assert_called_once()
-        assert mock_is_ready.call_count == 2
-
     def test_get_cached_cert_fingerprint_no_cert(self):
         with pytest.raises(ValueError, match="mTLS connection is not configured."):
             _agent_identity_utils.get_cached_cert_fingerprint(None)
