fix(auth): handle PermissionError on workload certificates to avoid startup hang and crash (googleapis#17568)

In sandboxed environments (such as when running the gcloud CLI within a
snap package manager sandbox), AppArmor rules can block reading the
workload credentials directory or certificate files, raising a
PermissionError.

Previously, `os.path.exists()` caught this PermissionError and returned
`False`. The library interpreted this as the files not being ready yet
(e.g. due to a startup race) and entered a 30-second retry loop before
failing with a RefreshError.

This change:
1. Updates `_is_certificate_file_ready` to use `os.stat` and propagate
`PermissionError`.
2. Catches `PermissionError` immediately during certificate lookup,
logging a warning and falling back to unbound tokens without retrying or
crashing.
3. Adds corresponding unit tests.

Fixes b/522957573

Author: nbayati

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

@@ -16,16 +16,15 @@
 
 import base64
 import hashlib
-import logging
 import os
 import re
+import stat
 import time
 from urllib.parse import quote, urlparse
+import warnings
 
 from google.auth import environment_vars, exceptions
 
-_LOGGER = logging.getLogger(__name__)
-
 CRYPTOGRAPHY_NOT_FOUND_ERROR = (
     "The cryptography library is required for certificate-based authentication."
     "Please install it with `pip install google-auth[cryptography]`."
@@ -58,8 +57,20 @@
 
 
 def _is_certificate_file_ready(path):
-    """Checks if a file exists and is not empty."""
-    return path and os.path.exists(path) and os.path.getsize(path) > 0
+    """Checks if a file exists, is a regular file, and is not empty."""
+    if not path:
+        return False
+    try:
+        # Check if the path points to a regular file and is not empty.
+        # stat.S_ISREG is used instead of os.path.isfile to avoid swallowing
+        # PermissionError exceptions, which the caller needs to propagate.
+        st = os.stat(path)
+        return stat.S_ISREG(st.st_mode) and st.st_size > 0
+    except PermissionError:
+        # Propagate PermissionError to let caller handle it (fail-fast or fallback)
+        raise
+    except OSError:
+        return False
 
 
 def get_agent_identity_certificate_path():
@@ -141,26 +152,28 @@ def get_agent_identity_certificate_path():
 
             # Log a warning on the first failed attempt to load the certificate file
             if not has_logged_cert_warning:
-                _LOGGER.warning(
-                    "Certificate file not ready at %s. Retrying until startup timeout (up to %s seconds total)...",
-                    target_path,
-                    _TOTAL_TIMEOUT,
+                warnings.warn(
+                    f"Certificate file not ready at {target_path}. Retrying until startup timeout (up to {_TOTAL_TIMEOUT} seconds total)..."
                 )
                 has_logged_cert_warning = True
 
+        except PermissionError as e:
+            warnings.warn(
+                f"Permission denied when accessing certificate config or certificate file: {e}. "
+                "Token binding protection cannot be enabled. Falling back to unbound tokens."
+            )
+            return None
         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,
+                warnings.warn(
+                    f"Certificate config file not found or incomplete: {e} (from "
+                    f"{environment_vars.GOOGLE_API_CERTIFICATE_CONFIG} environment variable). "
+                    f"Retrying until startup timeout (up to {_TOTAL_TIMEOUT} seconds total)..."
                 )
                 has_logged_config_warning = True
             pass
@@ -212,8 +225,15 @@ def get_and_parse_agent_identity_certificate():
     if not cert_path:
         return None
 
-    with open(cert_path, "rb") as cert_file:
-        cert_bytes = cert_file.read()
+    try:
+        with open(cert_path, "rb") as cert_file:
+            cert_bytes = cert_file.read()
+    except PermissionError as e:
+        warnings.warn(
+            f"Failed to read agent identity certificate file at {cert_path}: {e}. "
+            "Token binding protection cannot be enabled. Falling back to unbound tokens."
+        )
+        return None
 
     return parse_certificate(cert_bytes)
 

packages/google-auth/tests/compute_engine/test__mtls.py

@@ -78,13 +78,13 @@ def test__parse_mds_mode_invalid(monkeypatch):
         _mtls._parse_mds_mode()
 
 
-@mock.patch("os.path.exists")
+@mock.patch("google.auth.compute_engine._mtls.os.path.exists")
 def test__certs_exist_true(mock_exists, mock_mds_mtls_config):
     mock_exists.return_value = True
     assert _mtls._certs_exist(mock_mds_mtls_config) is True
 
 
-@mock.patch("os.path.exists")
+@mock.patch("google.auth.compute_engine._mtls.os.path.exists")
 def test__certs_exist_false(mock_exists, mock_mds_mtls_config):
     mock_exists.return_value = False
     assert _mtls._certs_exist(mock_mds_mtls_config) is False
@@ -101,7 +101,7 @@ def test__certs_exist_false(mock_exists, mock_mds_mtls_config):
         ("default", False, False),
     ],
 )
-@mock.patch("os.path.exists")
+@mock.patch("google.auth.compute_engine._mtls.os.path.exists")
 def test_should_use_mds_mtls(
     mock_exists, monkeypatch, mtls_mode, certs_exist, expected_result
 ):

packages/google-auth/tests/test_agent_identity_utils.py

@@ -65,6 +65,27 @@ def test_parse_certificate(self, mock_load_cert):
         mock_load_cert.assert_called_once_with(b"cert_bytes")
         assert result == mock_load_cert.return_value
 
+    @mock.patch("google.auth._agent_identity_utils.os.stat")
+    def test_is_certificate_file_ready_permission_error(self, mock_stat):
+        mock_stat.side_effect = PermissionError("Permission denied")
+        with pytest.raises(PermissionError):
+            _agent_identity_utils._is_certificate_file_ready("/path/to/cert")
+
+    @mock.patch("google.auth._agent_identity_utils.os.stat")
+    def test_is_certificate_file_ready_os_error(self, mock_stat):
+        mock_stat.side_effect = OSError("Not found")
+        # Should swallow the OSError and return False
+        result = _agent_identity_utils._is_certificate_file_ready("/path/to/cert")
+        assert result is False
+
+    @mock.patch("google.auth._agent_identity_utils.os.stat")
+    def test_is_certificate_file_ready_not_a_file(self, mock_stat):
+        import stat
+
+        mock_stat.return_value = mock.MagicMock(st_mode=stat.S_IFDIR, st_size=4096)
+        result = _agent_identity_utils._is_certificate_file_ready("/path/to/cert")
+        assert result is False
+
     def test__is_agent_identity_certificate_invalid(self):
         cert = _agent_identity_utils.parse_certificate(NON_AGENT_IDENTITY_CERT_BYTES)
         assert not _agent_identity_utils._is_agent_identity_certificate(cert)
@@ -268,7 +289,7 @@ def test_get_agent_identity_certificate_path_workstation_fail_fast(
         assert result is None
 
     @mock.patch("time.sleep")
-    @mock.patch("os.path.exists")
+    @mock.patch("google.auth._agent_identity_utils.os.path.exists")
     def test_get_agent_identity_certificate_path_cert_not_found(
         self, mock_exists, mock_sleep, tmpdir, monkeypatch
     ):
@@ -358,7 +379,7 @@ def test_get_agent_identity_certificate_path_workload_config_missing_cert_path(
         mock_sleep.assert_not_called()
 
     @mock.patch("time.sleep")
-    @mock.patch("os.path.exists")
+    @mock.patch("google.auth._agent_identity_utils.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
@@ -378,7 +399,7 @@ def test_get_agent_identity_certificate_path_no_config_but_has_well_known_dir(
         mock_sleep.assert_not_called()
 
     @mock.patch("time.sleep")
-    @mock.patch("os.path.exists")
+    @mock.patch("google.auth._agent_identity_utils.os.path.exists")
     def test_get_agent_identity_certificate_path_no_config_no_well_known_dir(
         self, mock_exists, mock_sleep, monkeypatch
     ):
@@ -396,7 +417,7 @@ def test_get_agent_identity_certificate_path_no_config_no_well_known_dir(
         mock_sleep.assert_not_called()
 
     @mock.patch("time.sleep")
-    @mock.patch("os.path.exists")
+    @mock.patch("google.auth._agent_identity_utils.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
@@ -415,7 +436,7 @@ def test_get_agent_identity_certificate_path_no_config_well_known_polling_succes
         assert mock_sleep.call_count == 1
 
     @mock.patch("time.sleep")
-    @mock.patch("os.path.exists")
+    @mock.patch("google.auth._agent_identity_utils.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
@@ -433,6 +454,45 @@ def test_get_agent_identity_certificate_path_no_config_well_known_polling_timeou
 
         assert mock_sleep.call_count == len(_agent_identity_utils._POLLING_INTERVALS)
 
+    @mock.patch("time.sleep")
+    @mock.patch("google.auth._agent_identity_utils._is_certificate_file_ready")
+    @mock.patch("google.auth._agent_identity_utils.os.path.exists")
+    def test_get_agent_identity_certificate_path_permission_error_well_known(
+        self, mock_exists, mock_is_ready, mock_sleep, monkeypatch
+    ):
+        monkeypatch.delenv(
+            environment_vars.GOOGLE_API_CERTIFICATE_CONFIG, raising=False
+        )
+        mock_exists.return_value = True
+        mock_is_ready.side_effect = PermissionError("Permission denied")
+
+        # It should fail-fast and return None immediately
+        result = _agent_identity_utils.get_agent_identity_certificate_path()
+        assert result is None
+        mock_sleep.assert_not_called()
+
+    @mock.patch("time.sleep")
+    @mock.patch("google.auth._agent_identity_utils.os.path.exists")
+    def test_get_agent_identity_certificate_path_permission_error_config(
+        self, mock_exists, mock_sleep, tmpdir, monkeypatch
+    ):
+        config_path = tmpdir.join("config.json")
+        monkeypatch.setenv(
+            environment_vars.GOOGLE_API_CERTIFICATE_CONFIG, str(config_path)
+        )
+        # Mock os.path.exists so ECP workstation fail-fast is not triggered
+        mock_exists.return_value = True
+
+        # Mocking open to raise PermissionError
+        mock_open = mock.mock_open()
+        mock_open.side_effect = PermissionError("Permission denied")
+
+        with mock.patch("builtins.open", mock_open):
+            result = _agent_identity_utils.get_agent_identity_certificate_path()
+
+        assert result is None
+        mock_sleep.assert_not_called()
+
     @mock.patch("google.auth._agent_identity_utils.get_agent_identity_certificate_path")
     def test_get_and_parse_agent_identity_certificate_opted_out(
         self, mock_get_path, monkeypatch
@@ -501,6 +561,23 @@ def test_get_and_parse_agent_identity_certificate_use_client_cert_invalid(
         assert result is None
         mock_get_path.assert_not_called()
 
+    @mock.patch("google.auth._agent_identity_utils.get_agent_identity_certificate_path")
+    def test_get_and_parse_agent_identity_certificate_file_read_error(
+        self, mock_get_path, monkeypatch
+    ):
+        monkeypatch.setenv(
+            environment_vars.GOOGLE_API_PREVENT_AGENT_TOKEN_SHARING_FOR_GCP_SERVICES,
+            "true",
+        )
+        mock_get_path.return_value = "/fake/cert.pem"
+        mock_open = mock.mock_open()
+        mock_open.side_effect = PermissionError("Permission denied")
+
+        with mock.patch("builtins.open", mock_open):
+            result = _agent_identity_utils.get_and_parse_agent_identity_certificate()
+
+        assert result is None
+
     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)