Implement updated design changes for RAB

google/auth/_constants.py

@@ -1,5 +1,5 @@
 """Shared constants."""
 
-_SERVICE_ACCOUNT_TRUST_BOUNDARY_LOOKUP_ENDPOINT = "https://iamcredentials.{universe_domain}/v1/projects/-/serviceAccounts/{service_account_email}/allowedLocations"
-_WORKFORCE_POOL_TRUST_BOUNDARY_LOOKUP_ENDPOINT = "https://iamcredentials.{universe_domain}/v1/locations/global/workforcePools/{pool_id}/allowedLocations"
-_WORKLOAD_IDENTITY_POOL_TRUST_BOUNDARY_LOOKUP_ENDPOINT = "https://iamcredentials.{universe_domain}/v1/projects/{project_number}/locations/global/workloadIdentityPools/{pool_id}/allowedLocations"
+_SERVICE_ACCOUNT_REGIONAL_ACCESS_BOUNDARY_LOOKUP_ENDPOINT = "https://iamcredentials.googleapis.com/v1/projects/-/serviceAccounts/{service_account_email}/allowedLocations"
+_WORKFORCE_POOL_REGIONAL_ACCESS_BOUNDARY_LOOKUP_ENDPOINT = "https://iamcredentials.googleapis.com/v1/locations/global/workforcePools/{pool_id}/allowedLocations"
+_WORKLOAD_IDENTITY_POOL_REGIONAL_ACCESS_BOUNDARY_LOOKUP_ENDPOINT = "https://iamcredentials.googleapis.com/v1/projects/{project_number}/locations/global/workloadIdentityPools/{pool_id}/allowedLocations"

google/auth/_helpers.py

@@ -314,8 +314,7 @@ def get_bool_from_env(variable_name, default=False):
     The environment variable is interpreted as a boolean with the following
     (case-insensitive) rules:
     - "true", "1" are considered true.
-    - "false", "0" are considered false.
-    Any other values will raise an exception.
+    - Any other value (or unset) is considered false.
 
     Args:
         variable_name (str): The name of the environment variable.
@@ -324,10 +323,6 @@ def get_bool_from_env(variable_name, default=False):
 
     Returns:
         bool: The boolean value of the environment variable.
-
-    Raises:
-        google.auth.exceptions.InvalidValue: If the environment variable is
-            set to a value that can not be interpreted as a boolean.
     """
     value = os.environ.get(variable_name)
 
@@ -338,14 +333,8 @@ def get_bool_from_env(variable_name, default=False):
 
     if value in ("true", "1"):
         return True
-    elif value in ("false", "0"):
-        return False
     else:
-        raise exceptions.InvalidValue(
-            'Environment variable "{}" must be one of "true", "false", "1", or "0".'.format(
-                variable_name
-            )
-        )
+        return False
 
 
 def is_python_3():

google/auth/_regional_access_boundary_utils.py

@@ -0,0 +1,104 @@
+"""Utilities for Regional Access Boundary management."""
+
+import datetime
+import threading
+
+from google.auth import _helpers
+from google.auth._default import _LOGGER
+
+
+# The default lifetime for a cached Regional Access Boundary.
+DEFAULT_REGIONAL_ACCESS_BOUNDARY_TTL = datetime.timedelta(hours=6)
+
+# The initial cooldown period for a failed Regional Access Boundary lookup.
+DEFAULT_REGIONAL_ACCESS_BOUNDARY_COOLDOWN = datetime.timedelta(minutes=15)
+
+# The maximum cooldown period for a failed Regional Access Boundary lookup.
+MAX_REGIONAL_ACCESS_BOUNDARY_COOLDOWN = datetime.timedelta(hours=6)
+
+
+class _RegionalAccessBoundaryRefreshThread(threading.Thread):
+    """Thread for background refreshing of the Regional Access Boundary."""
+
+    def __init__(self, credentials, request):
+        super(_RegionalAccessBoundaryRefreshThread, self).__init__()
+        self._credentials = credentials
+        self._request = request
+
+    def run(self):
+        """
+        Performs the Regional Access Boundary lookup and updates the credential's state.
+
+        This method is run in a separate thread. It delegates the actual lookup
+        to the credentials object's `_lookup_regional_access_boundary` method.
+        Based on the lookup's outcome (success or complete failure after retries),
+        it updates the credential's cached Regional Access Boundary information,
+        its expiry, its cooldown expiry, and its exponential cooldown duration.
+        """
+        regional_access_boundary_info = (
+            self._credentials._lookup_regional_access_boundary(self._request)
+        )
+
+        with self._credentials._stale_boundary_lock:  # Acquire the lock
+            if regional_access_boundary_info:
+                # On success, update the boundary and its expiry, and clear any cooldown.
+                self._credentials._regional_access_boundary = (
+                    regional_access_boundary_info
+                )
+                self._credentials._regional_access_boundary_expiry = (
+                    _helpers.utcnow() + DEFAULT_REGIONAL_ACCESS_BOUNDARY_TTL
+                )
+                self._credentials._regional_access_boundary_cooldown_expiry = None
+                # Reset the cooldown duration on success.
+                self._credentials._current_rab_cooldown_duration = (
+                    DEFAULT_REGIONAL_ACCESS_BOUNDARY_COOLDOWN
+                )
+                if _helpers.is_logging_enabled(_LOGGER):
+                    _LOGGER.debug(
+                        "Asynchronous Regional Access Boundary lookup successful."
+                    )
+            else:
+                # On complete failure, calculate the next exponential cooldown duration and set the cooldown expiry.
+                if _helpers.is_logging_enabled(_LOGGER):
+                    _LOGGER.warning(
+                        "Asynchronous Regional Access Boundary lookup failed. Entering cooldown."
+                    )
+                new_cooldown_duration = (
+                    self._credentials._current_rab_cooldown_duration * 2
+                )
+                self._credentials._current_rab_cooldown_duration = min(
+                    new_cooldown_duration, MAX_REGIONAL_ACCESS_BOUNDARY_COOLDOWN
+                )
+                self._credentials._regional_access_boundary_cooldown_expiry = (
+                    _helpers.utcnow() + self._credentials._current_rab_cooldown_duration
+                )
+                # If the proactive refresh failed, clear any existing expired RAB data.
+                # This ensures we don't continue using stale data.
+                self._credentials._regional_access_boundary = None
+                self._credentials._regional_access_boundary_expiry = None
+
+
+class _RegionalAccessBoundaryRefreshManager(object):
+    """Manages a thread for background refreshing of the Regional Access Boundary."""
+
+    def __init__(self):
+        self._lock = threading.Lock()
+        self._worker = None
+
+    def start_refresh(self, credentials, request):
+        """
+        Starts a background thread to refresh the Regional Access Boundary if one is not already running.
+
+        Args:
+            credentials (CredentialsWithRegionalAccessBoundary): The credentials
+                to refresh.
+            request (google.auth.transport.Request): The object used to make
+                HTTP requests.
+        """
+        with self._lock:
+            if self._worker and self._worker.is_alive():
+                # A refresh is already in progress.
+                return
+
+            self._worker = _RegionalAccessBoundaryRefreshThread(credentials, request)
+            self._worker.start()

google/auth/compute_engine/credentials.py

@@ -20,7 +20,11 @@
 """
 
 import datetime
+import logging
 
+_LOGGER = logging.getLogger(__name__)
+
+from google.auth import _constants
 from google.auth import _helpers
 from google.auth import credentials
 from google.auth import exceptions
@@ -30,16 +34,12 @@
 from google.auth.compute_engine import _metadata
 from google.oauth2 import _client
 
-_TRUST_BOUNDARY_LOOKUP_ENDPOINT = (
-    "https://iamcredentials.{}/v1/projects/-/serviceAccounts/{}/allowedLocations"
-)
-
 
 class Credentials(
     credentials.Scoped,
     credentials.CredentialsWithQuotaProject,
     credentials.CredentialsWithUniverseDomain,
-    credentials.CredentialsWithTrustBoundary,
+    credentials.CredentialsWithRegionalAccessBoundary,
 ):
     """Compute Engine Credentials.
 
@@ -66,7 +66,6 @@ def __init__(
         scopes=None,
         default_scopes=None,
         universe_domain=None,
-        trust_boundary=None,
     ):
         """
         Args:
@@ -82,7 +81,6 @@ def __init__(
                 provided or None, credential will attempt to fetch the value
                 from metadata server. If metadata server doesn't have universe
                 domain endpoint, then the default googleapis.com will be used.
-            trust_boundary (Mapping[str,str]): A credential trust boundary.
         """
         super(Credentials, self).__init__()
         self._service_account_email = service_account_email
@@ -93,7 +91,6 @@ def __init__(
         if universe_domain:
             self._universe_domain = universe_domain
             self._universe_domain_cached = True
-        self._trust_boundary = trust_boundary
 
     def _retrieve_info(self, request):
         """Retrieve information about the service account.
@@ -146,8 +143,8 @@ def _perform_refresh_token(self, request):
             new_exc = exceptions.RefreshError(caught_exc)
             raise new_exc from caught_exc
 
-    def _build_trust_boundary_lookup_url(self):
-        """Builds and returns the URL for the trust boundary lookup API for GCE."""
+    def _build_regional_access_boundary_lookup_url(self):
+        """Builds and returns the URL for the regional access boundary lookup API for GCE."""
         # If the service account email is 'default', we need to get the
         # actual email address from the metadata server.
         if self._service_account_email == "default":
@@ -157,24 +154,26 @@ def _build_trust_boundary_lookup_url(self):
             try:
                 info = _metadata.get_service_account_info(request, "default")
                 if not info or "email" not in info:
-                    raise exceptions.RefreshError(
+                    _LOGGER.error(
                         "Unexpected response from metadata server: "
-                        "service account info is missing 'email' field."
+                        "service account info is missing 'email' field. Cannot build Regional Access Boundary lookup URL."
                     )
+                    return None
                 self._service_account_email = info["email"]
 
             except exceptions.TransportError as e:
                 # If fetching the service account email fails due to a transport error,
-                # it means we cannot build the trust boundary lookup URL.
-                # Wrap this in a RefreshError so it's caught by _refresh_trust_boundary.
-                raise exceptions.RefreshError(
-                    "Failed to get service account email for trust boundary lookup: {}".format(
-                        e
-                    )
-                ) from e
+                # it means we cannot build the regional access boundary lookup URL.
+                _LOGGER.error(
+                    "Failed to get service account email to build Regional Access Boundary lookup URL: %s",
+                    e,
+                )
+                return None
 
-        return _TRUST_BOUNDARY_LOOKUP_ENDPOINT.format(
-            self.universe_domain, self.service_account_email
+        return (
+            _constants._SERVICE_ACCOUNT_REGIONAL_ACCESS_BOUNDARY_LOOKUP_ENDPOINT.format(
+                self.service_account_email
+            )
         )
 
     @property
@@ -211,57 +210,38 @@ def get_cred_info(self):
             "principal": self.service_account_email,
         }
 
-    @_helpers.copy_docstring(credentials.CredentialsWithQuotaProject)
-    def with_quota_project(self, quota_project_id):
+    def _make_copy(self):
         creds = self.__class__(
             service_account_email=self._service_account_email,
-            quota_project_id=quota_project_id,
+            quota_project_id=self._quota_project_id,
             scopes=self._scopes,
             default_scopes=self._default_scopes,
             universe_domain=self._universe_domain,
-            trust_boundary=self._trust_boundary,
         )
         creds._universe_domain_cached = self._universe_domain_cached
+        self._copy_regional_access_boundary_state(creds)
+        return creds
+
+    @_helpers.copy_docstring(credentials.CredentialsWithQuotaProject)
+    def with_quota_project(self, quota_project_id):
+        creds = self._make_copy()
+        creds._quota_project_id = quota_project_id
         return creds
 
     @_helpers.copy_docstring(credentials.Scoped)
     def with_scopes(self, scopes, default_scopes=None):
         # Compute Engine credentials can not be scoped (the metadata service
         # ignores the scopes parameter). App Engine, Cloud Run and Flex support
         # requesting scopes.
-        creds = self.__class__(
-            scopes=scopes,
-            default_scopes=default_scopes,
-            service_account_email=self._service_account_email,
-            quota_project_id=self._quota_project_id,
-            universe_domain=self._universe_domain,
-            trust_boundary=self._trust_boundary,
-        )
-        creds._universe_domain_cached = self._universe_domain_cached
+        creds = self._make_copy()
+        creds._scopes = scopes
+        creds._default_scopes = default_scopes
         return creds
 
     @_helpers.copy_docstring(credentials.CredentialsWithUniverseDomain)
     def with_universe_domain(self, universe_domain):
-        return self.__class__(
-            scopes=self._scopes,
-            default_scopes=self._default_scopes,
-            service_account_email=self._service_account_email,
-            quota_project_id=self._quota_project_id,
-            trust_boundary=self._trust_boundary,
-            universe_domain=universe_domain,
-        )
-
-    @_helpers.copy_docstring(credentials.CredentialsWithTrustBoundary)
-    def with_trust_boundary(self, trust_boundary):
-        creds = self.__class__(
-            service_account_email=self._service_account_email,
-            quota_project_id=self._quota_project_id,
-            scopes=self._scopes,
-            default_scopes=self._default_scopes,
-            universe_domain=self._universe_domain,
-            trust_boundary=trust_boundary,
-        )
-        creds._universe_domain_cached = self._universe_domain_cached
+        creds = self._make_copy()
+        creds._universe_domain = universe_domain
         return creds