fix: Trust boundary meta header renaming and using the schema from ba…

…ckend team. (#1384) * fix: rename the trust boundary metaheader into * fix comments
googleapis · Sep 21, 2023 · 2503d4a · 2503d4a
1 parent bd25e6a
commit 2503d4a
Show file tree

Hide file tree

Showing 7 changed files with 54 additions and 37 deletions.
diff --git a/google/auth/credentials.py b/google/auth/credentials.py
@@ -52,8 +52,9 @@ def __init__(self):
         self._quota_project_id = None
         """Optional[str]: Project to use for quota and billing purposes."""
         self._trust_boundary = None
-        """Optional[str]: Encoded string representation of credentials trust
-        boundary."""
+        """Optional[dict]: Cache of a trust boundary response which has a list
+        of allowed regions and an encoded string representation of credentials
+        trust boundary."""
         self._universe_domain = "googleapis.com"
         """Optional[str]: The universe domain value, default is googleapis.com
         """
@@ -135,8 +136,21 @@ def apply(self, headers, token=None):
         headers["authorization"] = "Bearer {}".format(
             _helpers.from_bytes(token or self.token)
         )
+        """Trust boundary value will be a cached value from global lookup.
+
+        The response of trust boundary will be a list of regions and a hex
+        encoded representation.
+
+        An example of global lookup response:
+        {
+          "locations": [
+            "us-central1", "us-east1", "europe-west1", "asia-east1"
+          ]
+          "encoded_locations": "0xA30"
+        }
+        """
         if self._trust_boundary is not None:
-            headers["x-identity-trust-boundary"] = self._trust_boundary
+            headers["x-allowed-locations"] = self._trust_boundary["encoded_locations"]
         if self.quota_project_id:
             headers["x-goog-user-project"] = self.quota_project_id
 

diff --git a/google/auth/external_account.py b/google/auth/external_account.py
@@ -132,7 +132,10 @@ def __init__(
         self._default_scopes = default_scopes
         self._workforce_pool_user_project = workforce_pool_user_project
         self._universe_domain = universe_domain or _DEFAULT_UNIVERSE_DOMAIN
-        self._trust_boundary = "0"  # expose a placeholder trust boundary value.
+        self._trust_boundary = {
+            "locations": [],
+            "encoded_locations": "0x0",
+        }  # expose a placeholder trust boundary value.
 
         if self._client_id:
             self._client_auth = utils.ClientAuthentication(

diff --git a/google/oauth2/service_account.py b/google/oauth2/service_account.py
@@ -196,7 +196,7 @@ def __init__(
             self._additional_claims = additional_claims
         else:
             self._additional_claims = {}
-        self._trust_boundary = "0"
+        self._trust_boundary = {"locations": [], "encoded_locations": "0x0"}
 
     @classmethod
     def _from_signer_and_info(cls, signer, info, **kwargs):

diff --git a/tests/test_aws.py b/tests/test_aws.py
@@ -1969,7 +1969,7 @@ def test_refresh_success_with_impersonation_ignore_default_scopes(
             "authorization": "Bearer {}".format(self.SUCCESS_RESPONSE["access_token"]),
             "x-goog-user-project": QUOTA_PROJECT_ID,
             "x-goog-api-client": IMPERSONATE_ACCESS_TOKEN_REQUEST_METRICS_HEADER_VALUE,
-            "x-identity-trust-boundary": "0",
+            "x-allowed-locations": "0x0",
         }
         impersonation_request_data = {
             "delegates": None,
@@ -2066,7 +2066,7 @@ def test_refresh_success_with_impersonation_use_default_scopes(
             "authorization": "Bearer {}".format(self.SUCCESS_RESPONSE["access_token"]),
             "x-goog-user-project": QUOTA_PROJECT_ID,
             "x-goog-api-client": IMPERSONATE_ACCESS_TOKEN_REQUEST_METRICS_HEADER_VALUE,
-            "x-identity-trust-boundary": "0",
+            "x-allowed-locations": "0x0",
         }
         impersonation_request_data = {
             "delegates": None,

diff --git a/tests/test_credentials.py b/tests/test_credentials.py
@@ -81,7 +81,7 @@ def test_before_request():
     assert credentials.valid
     assert credentials.token == "token"
     assert headers["authorization"] == "Bearer token"
-    assert "x-identity-trust-boundary" not in headers
+    assert "x-allowed-locations" not in headers
 
     request = "token2"
     headers = {}
@@ -91,13 +91,13 @@ def test_before_request():
     assert credentials.valid
     assert credentials.token == "token"
     assert headers["authorization"] == "Bearer token"
-    assert "x-identity-trust-boundary" not in headers
+    assert "x-allowed-locations" not in headers
 
 
 def test_before_request_with_trust_boundary():
-    DUMMY_BOUNDARY = "00110101"
+    DUMMY_BOUNDARY = "0xA30"
     credentials = CredentialsImpl()
-    credentials._trust_boundary = DUMMY_BOUNDARY
+    credentials._trust_boundary = {"locations": [], "encoded_locations": DUMMY_BOUNDARY}
     request = "token"
     headers = {}
 
@@ -106,7 +106,7 @@ def test_before_request_with_trust_boundary():
     assert credentials.valid
     assert credentials.token == "token"
     assert headers["authorization"] == "Bearer token"
-    assert headers["x-identity-trust-boundary"] == DUMMY_BOUNDARY
+    assert headers["x-allowed-locations"] == DUMMY_BOUNDARY
 
     request = "token2"
     headers = {}
@@ -116,7 +116,7 @@ def test_before_request_with_trust_boundary():
     assert credentials.valid
     assert credentials.token == "token"
     assert headers["authorization"] == "Bearer token"
-    assert headers["x-identity-trust-boundary"] == DUMMY_BOUNDARY
+    assert headers["x-allowed-locations"] == DUMMY_BOUNDARY
 
 
 def test_before_request_metrics():

diff --git a/tests/test_external_account.py b/tests/test_external_account.py
@@ -833,7 +833,7 @@ def test_refresh_impersonation_without_client_auth_success(
             "Content-Type": "application/json",
             "authorization": "Bearer {}".format(token_response["access_token"]),
             "x-goog-api-client": IMPERSONATE_ACCESS_TOKEN_REQUEST_METRICS_HEADER_VALUE,
-            "x-identity-trust-boundary": "0",
+            "x-allowed-locations": "0x0",
         }
         impersonation_request_data = {
             "delegates": None,
@@ -915,7 +915,7 @@ def test_refresh_workforce_impersonation_without_client_auth_success(
             "Content-Type": "application/json",
             "authorization": "Bearer {}".format(token_response["access_token"]),
             "x-goog-api-client": IMPERSONATE_ACCESS_TOKEN_REQUEST_METRICS_HEADER_VALUE,
-            "x-identity-trust-boundary": "0",
+            "x-allowed-locations": "0x0",
         }
         impersonation_request_data = {
             "delegates": None,
@@ -1134,7 +1134,7 @@ def test_refresh_impersonation_with_client_auth_success_ignore_default_scopes(
             "Content-Type": "application/json",
             "authorization": "Bearer {}".format(token_response["access_token"]),
             "x-goog-api-client": IMPERSONATE_ACCESS_TOKEN_REQUEST_METRICS_HEADER_VALUE,
-            "x-identity-trust-boundary": "0",
+            "x-allowed-locations": "0x0",
         }
         impersonation_request_data = {
             "delegates": None,
@@ -1218,7 +1218,7 @@ def test_refresh_impersonation_with_client_auth_success_use_default_scopes(
             "Content-Type": "application/json",
             "authorization": "Bearer {}".format(token_response["access_token"]),
             "x-goog-api-client": IMPERSONATE_ACCESS_TOKEN_REQUEST_METRICS_HEADER_VALUE,
-            "x-identity-trust-boundary": "0",
+            "x-allowed-locations": "0x0",
         }
         impersonation_request_data = {
             "delegates": None,
@@ -1274,7 +1274,7 @@ def test_apply_without_quota_project_id(self):
 
         assert headers == {
             "authorization": "Bearer {}".format(self.SUCCESS_RESPONSE["access_token"]),
-            "x-identity-trust-boundary": "0",
+            "x-allowed-locations": "0x0",
         }
 
     def test_apply_workforce_without_quota_project_id(self):
@@ -1291,7 +1291,7 @@ def test_apply_workforce_without_quota_project_id(self):
 
         assert headers == {
             "authorization": "Bearer {}".format(self.SUCCESS_RESPONSE["access_token"]),
-            "x-identity-trust-boundary": "0",
+            "x-allowed-locations": "0x0",
         }
 
     def test_apply_impersonation_without_quota_project_id(self):
@@ -1323,7 +1323,7 @@ def test_apply_impersonation_without_quota_project_id(self):
 
         assert headers == {
             "authorization": "Bearer {}".format(impersonation_response["accessToken"]),
-            "x-identity-trust-boundary": "0",
+            "x-allowed-locations": "0x0",
         }
 
     def test_apply_with_quota_project_id(self):
@@ -1340,7 +1340,7 @@ def test_apply_with_quota_project_id(self):
             "other": "header-value",
             "authorization": "Bearer {}".format(self.SUCCESS_RESPONSE["access_token"]),
             "x-goog-user-project": self.QUOTA_PROJECT_ID,
-            "x-identity-trust-boundary": "0",
+            "x-allowed-locations": "0x0",
         }
 
     def test_apply_impersonation_with_quota_project_id(self):
@@ -1375,7 +1375,7 @@ def test_apply_impersonation_with_quota_project_id(self):
             "other": "header-value",
             "authorization": "Bearer {}".format(impersonation_response["accessToken"]),
             "x-goog-user-project": self.QUOTA_PROJECT_ID,
-            "x-identity-trust-boundary": "0",
+            "x-allowed-locations": "0x0",
         }
 
     def test_before_request(self):
@@ -1391,7 +1391,7 @@ def test_before_request(self):
         assert headers == {
             "other": "header-value",
             "authorization": "Bearer {}".format(self.SUCCESS_RESPONSE["access_token"]),
-            "x-identity-trust-boundary": "0",
+            "x-allowed-locations": "0x0",
         }
 
         # Second call shouldn't call refresh.
@@ -1400,7 +1400,7 @@ def test_before_request(self):
         assert headers == {
             "other": "header-value",
             "authorization": "Bearer {}".format(self.SUCCESS_RESPONSE["access_token"]),
-            "x-identity-trust-boundary": "0",
+            "x-allowed-locations": "0x0",
         }
 
     def test_before_request_workforce(self):
@@ -1418,7 +1418,7 @@ def test_before_request_workforce(self):
         assert headers == {
             "other": "header-value",
             "authorization": "Bearer {}".format(self.SUCCESS_RESPONSE["access_token"]),
-            "x-identity-trust-boundary": "0",
+            "x-allowed-locations": "0x0",
         }
 
         # Second call shouldn't call refresh.
@@ -1427,7 +1427,7 @@ def test_before_request_workforce(self):
         assert headers == {
             "other": "header-value",
             "authorization": "Bearer {}".format(self.SUCCESS_RESPONSE["access_token"]),
-            "x-identity-trust-boundary": "0",
+            "x-allowed-locations": "0x0",
         }
 
     def test_before_request_impersonation(self):
@@ -1458,7 +1458,7 @@ def test_before_request_impersonation(self):
         assert headers == {
             "other": "header-value",
             "authorization": "Bearer {}".format(impersonation_response["accessToken"]),
-            "x-identity-trust-boundary": "0",
+            "x-allowed-locations": "0x0",
         }
 
         # Second call shouldn't call refresh.
@@ -1467,7 +1467,7 @@ def test_before_request_impersonation(self):
         assert headers == {
             "other": "header-value",
             "authorization": "Bearer {}".format(impersonation_response["accessToken"]),
-            "x-identity-trust-boundary": "0",
+            "x-allowed-locations": "0x0",
         }
 
     @mock.patch("google.auth._helpers.utcnow")
@@ -1495,7 +1495,7 @@ def test_before_request_expired(self, utcnow):
         # Cached token should be used.
         assert headers == {
             "authorization": "Bearer token",
-            "x-identity-trust-boundary": "0",
+            "x-allowed-locations": "0x0",
         }
 
         # Next call should simulate 1 second passed.
@@ -1509,7 +1509,7 @@ def test_before_request_expired(self, utcnow):
         # New token should be retrieved.
         assert headers == {
             "authorization": "Bearer {}".format(self.SUCCESS_RESPONSE["access_token"]),
-            "x-identity-trust-boundary": "0",
+            "x-allowed-locations": "0x0",
         }
 
     @mock.patch("google.auth._helpers.utcnow")
@@ -1552,7 +1552,7 @@ def test_before_request_impersonation_expired(self, utcnow):
         # Cached token should be used.
         assert headers == {
             "authorization": "Bearer token",
-            "x-identity-trust-boundary": "0",
+            "x-allowed-locations": "0x0",
         }
 
         # Next call should simulate 1 second passed. This will trigger the expiration
@@ -1567,7 +1567,7 @@ def test_before_request_impersonation_expired(self, utcnow):
         # New token should be retrieved.
         assert headers == {
             "authorization": "Bearer {}".format(impersonation_response["accessToken"]),
-            "x-identity-trust-boundary": "0",
+            "x-allowed-locations": "0x0",
         }
 
     @pytest.mark.parametrize(
@@ -1666,7 +1666,7 @@ def test_get_project_id_cloud_resource_manager_success(
             "x-goog-user-project": self.QUOTA_PROJECT_ID,
             "authorization": "Bearer {}".format(token_response["access_token"]),
             "x-goog-api-client": IMPERSONATE_ACCESS_TOKEN_REQUEST_METRICS_HEADER_VALUE,
-            "x-identity-trust-boundary": "0",
+            "x-allowed-locations": "0x0",
         }
         impersonation_request_data = {
             "delegates": None,
@@ -1720,7 +1720,7 @@ def test_get_project_id_cloud_resource_manager_success(
                 "authorization": "Bearer {}".format(
                     impersonation_response["accessToken"]
                 ),
-                "x-identity-trust-boundary": "0",
+                "x-allowed-locations": "0x0",
             },
         )
 
@@ -1792,7 +1792,7 @@ def test_workforce_pool_get_project_id_cloud_resource_manager_success(
                 "authorization": "Bearer {}".format(
                     self.SUCCESS_RESPONSE["access_token"]
                 ),
-                "x-identity-trust-boundary": "0",
+                "x-allowed-locations": "0x0",
             },
         )
 
@@ -1842,7 +1842,7 @@ def test_refresh_impersonation_with_lifetime(
             "Content-Type": "application/json",
             "authorization": "Bearer {}".format(token_response["access_token"]),
             "x-goog-api-client": IMPERSONATE_ACCESS_TOKEN_REQUEST_METRICS_HEADER_VALUE,
-            "x-identity-trust-boundary": "0",
+            "x-allowed-locations": "0x0",
         }
         impersonation_request_data = {
             "delegates": None,

diff --git a/tests/test_identity_pool.py b/tests/test_identity_pool.py
@@ -319,7 +319,7 @@ def assert_underlying_credentials_refresh(
                 "Content-Type": "application/json",
                 "authorization": "Bearer {}".format(token_response["access_token"]),
                 "x-goog-api-client": metrics_header_value,
-                "x-identity-trust-boundary": "0",
+                "x-allowed-locations": "0x0",
             }
             impersonation_request_data = {
                 "delegates": None,