feat: Add trust boundary support for service accounts and impersonation. (#1778)

* feat: adds trust boundary lookup support for SA and impersonated credentials

* Add feature flag, fix ud bug, and update no-op response.

* Add _build_trust_boundary_lookup_url to external account.

* Implement additional unit tests for the trust boundary

* implement trust boundary in compute_engine to support GCE instances.

* Fix failing unit test, and change acceptable values for the env variable

* Add unit tests for gce trust boundary.

* Use no op method instead of directly comparing values

* fix a typo in calling the method noop

* Add x-allowed-location header to all IAM requests.

* Support self-signed jwt and refactor refresh to handle refreshing trust boundary in the base class.

* Fix failing unit tests

* Revert changes to external account class file.

* Additional unit tests and update some old ones.

* Revert changes to idtoken

* Revert the self signed jwt token workaround

* revert idtoken and jwt trust boundary tests.

* Fix failing github check

* Fix failing unit tests in compute engine

* fix linter issues

* Fix linter issues

* remove trust boundary related code from idtoken as it is not a supported credential type

* remove unused line in test

Author: nbayati

google/auth/_helpers.py

@@ -21,6 +21,7 @@
 import hashlib
 import json
 import logging
+import os
 import sys
 from typing import Any, Dict, Mapping, Optional, Union
 import urllib
@@ -287,6 +288,46 @@ def unpadded_urlsafe_b64encode(value):
     return base64.urlsafe_b64encode(value).rstrip(b"=")
 
 
+def get_bool_from_env(variable_name, default=False):
+    """Gets a boolean value from an environment variable.
+
+    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.
+
+    Args:
+        variable_name (str): The name of the environment variable.
+        default (bool): The default value if the environment variable is not
+            set.
+
+    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)
+
+    if value is None:
+        return default
+
+    value = value.lower()
+
+    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
+            )
+        )
+
+
 def is_python_3():
     """Check if the Python interpreter is Python 2 or 3.
 

google/auth/compute_engine/credentials.py

@@ -30,11 +30,16 @@
 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,
 ):
     """Compute Engine Credentials.
 
@@ -61,6 +66,7 @@ def __init__(
         scopes=None,
         default_scopes=None,
         universe_domain=None,
+        trust_boundary=None,
     ):
         """
         Args:
@@ -76,6 +82,7 @@ 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
@@ -86,6 +93,7 @@ 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.
@@ -100,16 +108,22 @@ def _retrieve_info(self, request):
             request, service_account=self._service_account_email
         )
 
+        if not info or "email" not in info:
+            raise exceptions.RefreshError(
+                "Unexpected response from metadata server: "
+                "service account info is missing 'email' field."
+            )
+
         self._service_account_email = info["email"]
 
         # Don't override scopes requested by the user.
         if self._scopes is None:
-            self._scopes = info["scopes"]
+            self._scopes = info.get("scopes")
 
     def _metric_header_for_usage(self):
         return metrics.CRED_TYPE_SA_MDS
 
-    def refresh(self, request):
+    def _refresh_token(self, request):
         """Refresh the access token and scopes.
 
         Args:
@@ -132,6 +146,37 @@ def refresh(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."""
+        # 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":
+            from google.auth.transport import requests as google_auth_requests
+
+            request = google_auth_requests.Request()
+            try:
+                info = _metadata.get_service_account_info(request, "default")
+                if not info or "email" not in info:
+                    raise exceptions.RefreshError(
+                        "Unexpected response from metadata server: "
+                        "service account info is missing 'email' field."
+                    )
+                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
+
+        return _TRUST_BOUNDARY_LOOKUP_ENDPOINT.format(
+            self.universe_domain, self.service_account_email
+        )
+
     @property
     def service_account_email(self):
         """The service account email.
@@ -173,8 +218,9 @@ def with_quota_project(self, quota_project_id):
             quota_project_id=quota_project_id,
             scopes=self._scopes,
             default_scopes=self._default_scopes,
+            universe_domain=self._universe_domain,
+            trust_boundary=self._trust_boundary,
         )
-        creds._universe_domain = self._universe_domain
         creds._universe_domain_cached = self._universe_domain_cached
         return creds
 
@@ -188,8 +234,9 @@ def with_scopes(self, scopes, default_scopes=None):
             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 = self._universe_domain
         creds._universe_domain_cached = self._universe_domain_cached
         return creds
 
@@ -200,9 +247,23 @@ def with_universe_domain(self, universe_domain):
             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
+        return creds
+
 
 _DEFAULT_TOKEN_LIFETIME_SECS = 3600  # 1 hour in seconds
 _DEFAULT_TOKEN_URI = "https://www.googleapis.com/oauth2/v4/token"
@@ -275,7 +336,7 @@ def __init__(
 
         if use_metadata_identity_endpoint:
             if token_uri or additional_claims or service_account_email or signer:
-                raise exceptions.MalformedError(
+                raise ValueError(
                     "If use_metadata_identity_endpoint is set, token_uri, "
                     "additional_claims, service_account_email, signer arguments"
                     " must not be set"
@@ -366,7 +427,7 @@ def with_token_uri(self, token_uri):
         # since the signer is already instantiated,
         # the request is not needed
         if self._use_metadata_identity_endpoint:
-            raise exceptions.MalformedError(
+            raise ValueError(
                 "If use_metadata_identity_endpoint is set, token_uri" " must not be set"
             )
         else: