feat: adding meta header for trust boundary (#1334)

* feat: adding meta header for trust boundary

* fixing lint

* adding trust_boundary parameter for 3PI init

* change inject header to kebab case and the value to a reasonable value

google/auth/credentials.py

@@ -54,6 +54,9 @@ def __init__(self):
  If this is None, the token is assumed to never expire."""
  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."""
 
  @property
  def expired(self):
@@ -127,6 +130,8 @@ def apply(self, headers, token=None):
  headers["authorization"] = "Bearer {}".format(
  _helpers.from_bytes(token or self.token)
  )
+ if self._trust_boundary is not None:
+ headers["x-identity-trust-boundary"] = self._trust_boundary
  if self.quota_project_id:
  headers["x-goog-user-project"] = self.quota_project_id
 

google/auth/external_account.py

@@ -86,6 +86,7 @@ def __init__(
  default_scopes=None,
  workforce_pool_user_project=None,
  universe_domain=_DEFAULT_UNIVERSE_DOMAIN,
+ trust_boundary=None,
  ):
  """Instantiates an external account credentials object.
 
@@ -111,6 +112,7 @@ def __init__(
  billing/quota.
  universe_domain (str): The universe domain. The default universe
  domain is googleapis.com.
+ trust_boundary (str): String representation of trust boundary meta.
  Raises:
  google.auth.exceptions.RefreshError: If the generateAccessToken
  endpoint returned an error.
@@ -132,6 +134,7 @@ 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.
 
  if self._client_id:
  self._client_auth = utils.ClientAuthentication(

google/oauth2/credentials.py

@@ -84,6 +84,7 @@ def __init__(
  refresh_handler=None,
  enable_reauth_refresh=False,
  granted_scopes=None,
+ trust_boundary=None,
  ):
  """
  Args:
@@ -141,6 +142,7 @@ def __init__(
  self._rapt_token = rapt_token
  self.refresh_handler = refresh_handler
  self._enable_reauth_refresh = enable_reauth_refresh
+ self._trust_boundary = trust_boundary
 
  def __getstate__(self):
  """A __getstate__ method must exist for the __setstate__ to be called
@@ -171,6 +173,7 @@ def __setstate__(self, d):
  self._quota_project_id = d.get("_quota_project_id")
  self._rapt_token = d.get("_rapt_token")
  self._enable_reauth_refresh = d.get("_enable_reauth_refresh")
+ self._trust_boundary = d.get("_trust_boundary")
  # The refresh_handler setter should be used to repopulate this.
  self._refresh_handler = None
 
@@ -268,6 +271,7 @@ def with_quota_project(self, quota_project_id):
  quota_project_id=quota_project_id,
  rapt_token=self.rapt_token,
  enable_reauth_refresh=self._enable_reauth_refresh,
+ trust_boundary=self._trust_boundary,
  )
 
  @_helpers.copy_docstring(credentials.CredentialsWithTokenUri)
@@ -286,6 +290,7 @@ def with_token_uri(self, token_uri):
  quota_project_id=self.quota_project_id,
  rapt_token=self.rapt_token,
  enable_reauth_refresh=self._enable_reauth_refresh,
+ trust_boundary=self._trust_boundary,
  )
 
  def _metric_header_for_usage(self):
@@ -421,6 +426,7 @@ def from_authorized_user_info(cls, info, scopes=None):
  quota_project_id=info.get("quota_project_id"), # may not exist
  expiry=expiry,
  rapt_token=info.get("rapt_token"), # may not exist
+ trust_boundary=info.get("trust_boundary"), # may not exist
  )
 
  @classmethod

google/oauth2/service_account.py

@@ -140,6 +140,7 @@ def __init__(
  additional_claims=None,
  always_use_jwt_access=False,
  universe_domain=_DEFAULT_UNIVERSE_DOMAIN,
+ trust_boundary=None,
  ):
  """
  Args:
@@ -163,6 +164,7 @@ def __init__(
  universe_domain (str): The universe domain. The default
  universe domain is googleapis.com. For default value self
  signed jwt is used for token refresh.
+ trust_boundary (str): String representation of trust boundary meta.
 
  .. note:: Typically one of the helper constructors
  :meth:`from_service_account_file` or
@@ -194,6 +196,7 @@ def __init__(
  self._additional_claims = additional_claims
  else:
  self._additional_claims = {}
+ self._trust_boundary = "0"
 
  @classmethod
  def _from_signer_and_info(cls, signer, info, **kwargs):
@@ -217,6 +220,7 @@ def _from_signer_and_info(cls, signer, info, **kwargs):
  token_uri=info["token_uri"],
  project_id=info.get("project_id"),
  universe_domain=info.get("universe_domain", _DEFAULT_UNIVERSE_DOMAIN),
+ trust_boundary=info.get("trust_boundary"),
  **kwargs
  )
 

tests/test_aws.py

@@ -1969,6 +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",
  }
  impersonation_request_data = {
  "delegates": None,
@@ -2065,6 +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",
  }
  impersonation_request_data = {
  "delegates": None,

tests/test_credentials.py

@@ -80,6 +80,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
 
  request = "token2"
  headers = {}
@@ -89,6 +90,32 @@ def test_before_request():
  assert credentials.valid
  assert credentials.token == "token"
  assert headers["authorization"] == "Bearer token"
+ assert "x-identity-trust-boundary" not in headers
+
+
+def test_before_request_with_trust_boundary():
+ DUMMY_BOUNDARY = "00110101"
+ credentials = CredentialsImpl()
+ credentials._trust_boundary = DUMMY_BOUNDARY
+ request = "token"
+ headers = {}
+
+ # First call should call refresh, setting the token.
+ credentials.before_request(request, "http://example.com", "GET", headers)
+ assert credentials.valid
+ assert credentials.token == "token"
+ assert headers["authorization"] == "Bearer token"
+ assert headers["x-identity-trust-boundary"] == DUMMY_BOUNDARY
+
+ request = "token2"
+ headers = {}
+
+ # Second call shouldn't call refresh.
+ credentials.before_request(request, "http://example.com", "GET", headers)
+ assert credentials.valid
+ assert credentials.token == "token"
+ assert headers["authorization"] == "Bearer token"
+ assert headers["x-identity-trust-boundary"] == DUMMY_BOUNDARY
 
 
 def test_before_request_metrics():

tests/test_external_account.py

@@ -825,6 +825,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",
  }
  impersonation_request_data = {
  "delegates": None,
@@ -906,6 +907,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",
  }
  impersonation_request_data = {
  "delegates": None,
@@ -1124,6 +1126,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",
  }
  impersonation_request_data = {
  "delegates": None,
@@ -1207,6 +1210,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",
  }
  impersonation_request_data = {
  "delegates": None,
@@ -1261,7 +1265,8 @@ def test_apply_without_quota_project_id(self):
  credentials.apply(headers)
 
  assert headers == {
- "authorization": "Bearer {}".format(self.SUCCESS_RESPONSE["access_token"])
+ "authorization": "Bearer {}".format(self.SUCCESS_RESPONSE["access_token"]),
+ "x-identity-trust-boundary": "0",
  }
 
  def test_apply_workforce_without_quota_project_id(self):
@@ -1277,7 +1282,8 @@ def test_apply_workforce_without_quota_project_id(self):
  credentials.apply(headers)
 
  assert headers == {
- "authorization": "Bearer {}".format(self.SUCCESS_RESPONSE["access_token"])
+ "authorization": "Bearer {}".format(self.SUCCESS_RESPONSE["access_token"]),
+ "x-identity-trust-boundary": "0",
  }
 
  def test_apply_impersonation_without_quota_project_id(self):
@@ -1308,7 +1314,8 @@ def test_apply_impersonation_without_quota_project_id(self):
  credentials.apply(headers)
 
  assert headers == {
- "authorization": "Bearer {}".format(impersonation_response["accessToken"])
+ "authorization": "Bearer {}".format(impersonation_response["accessToken"]),
+ "x-identity-trust-boundary": "0",
  }
 
  def test_apply_with_quota_project_id(self):
@@ -1325,6 +1332,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",
  }
 
  def test_apply_impersonation_with_quota_project_id(self):
@@ -1359,6 +1367,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",
  }
 
  def test_before_request(self):
@@ -1374,6 +1383,7 @@ def test_before_request(self):
  assert headers == {
  "other": "header-value",
  "authorization": "Bearer {}".format(self.SUCCESS_RESPONSE["access_token"]),
+ "x-identity-trust-boundary": "0",
  }
 
  # Second call shouldn't call refresh.
@@ -1382,6 +1392,7 @@ def test_before_request(self):
  assert headers == {
  "other": "header-value",
  "authorization": "Bearer {}".format(self.SUCCESS_RESPONSE["access_token"]),
+ "x-identity-trust-boundary": "0",
  }
 
  def test_before_request_workforce(self):
@@ -1399,6 +1410,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",
  }
 
  # Second call shouldn't call refresh.
@@ -1407,6 +1419,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",
  }
 
  def test_before_request_impersonation(self):
@@ -1437,6 +1450,7 @@ def test_before_request_impersonation(self):
  assert headers == {
  "other": "header-value",
  "authorization": "Bearer {}".format(impersonation_response["accessToken"]),
+ "x-identity-trust-boundary": "0",
  }
 
  # Second call shouldn't call refresh.
@@ -1445,6 +1459,7 @@ def test_before_request_impersonation(self):
  assert headers == {
  "other": "header-value",
  "authorization": "Bearer {}".format(impersonation_response["accessToken"]),
+ "x-identity-trust-boundary": "0",
  }
 
  @mock.patch("google.auth._helpers.utcnow")
@@ -1470,7 +1485,10 @@ def test_before_request_expired(self, utcnow):
  credentials.before_request(request, "POST", "https://example.com/api", headers)
 
  # Cached token should be used.
- assert headers == {"authorization": "Bearer token"}
+ assert headers == {
+ "authorization": "Bearer token",
+ "x-identity-trust-boundary": "0",
+ }
 
  # Next call should simulate 1 second passed.
  utcnow.return_value = datetime.datetime.min + datetime.timedelta(seconds=1)
@@ -1482,7 +1500,8 @@ def test_before_request_expired(self, utcnow):
 
  # New token should be retrieved.
  assert headers == {
- "authorization": "Bearer {}".format(self.SUCCESS_RESPONSE["access_token"])
+ "authorization": "Bearer {}".format(self.SUCCESS_RESPONSE["access_token"]),
+ "x-identity-trust-boundary": "0",
  }
 
  @mock.patch("google.auth._helpers.utcnow")
@@ -1523,7 +1542,10 @@ def test_before_request_impersonation_expired(self, utcnow):
  credentials.before_request(request, "POST", "https://example.com/api", headers)
 
  # Cached token should be used.
- assert headers == {"authorization": "Bearer token"}
+ assert headers == {
+ "authorization": "Bearer token",
+ "x-identity-trust-boundary": "0",
+ }
 
  # Next call should simulate 1 second passed. This will trigger the expiration
  # threshold.
@@ -1536,7 +1558,8 @@ def test_before_request_impersonation_expired(self, utcnow):
 
  # New token should be retrieved.
  assert headers == {
- "authorization": "Bearer {}".format(impersonation_response["accessToken"])
+ "authorization": "Bearer {}".format(impersonation_response["accessToken"]),
+ "x-identity-trust-boundary": "0",
  }
 
  @pytest.mark.parametrize(
@@ -1635,6 +1658,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",
  }
  impersonation_request_data = {
  "delegates": None,
@@ -1688,6 +1712,7 @@ def test_get_project_id_cloud_resource_manager_success(
  "authorization": "Bearer {}".format(
  impersonation_response["accessToken"]
  ),
+ "x-identity-trust-boundary": "0",
  },
  )
 
@@ -1759,6 +1784,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",
  },
  )
 
@@ -1808,6 +1834,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",
  }
  impersonation_request_data = {
  "delegates": None,

tests/test_identity_pool.py

@@ -319,6 +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",
  }
  impersonation_request_data = {
  "delegates": None,