feat: define `google.auth.downscoped.Credentials` class (#801)
* feat: define `google.auth.downscoped.Credentials` class
This is based on [Downscoping with Credential Access Boundaries](https://cloud.google.com/iam/docs/downscoping-short-lived-credentials).
The new credentials are initialized mainly using elevated source
credentials and a `google.auth.downscoped.CredentialAccessBoundary`
instance.
The credentials will then get access tokens from the source
credentials and exchange them via the GCP STS token exchange
endpoint using the provided credentials access boundary rules
for downscoped access tokens.
The new credentials will inherit the source credentials' scopes
but the scopes are not exposed as we cannot always determine the
scopes form the source credentials.
* Fixes typos in comments.
* Addresses review comments.
* Moves all constants in the test file to module scope.
diff --git a/google/auth/downscoped.py b/google/auth/downscoped.py
index beea50e..800f289 100644
--- a/google/auth/downscoped.py
+++ b/google/auth/downscoped.py
@@ -48,9 +48,24 @@
.. _Downscoping with Credential Access Boundaries: https://cloud.google.com/iam/docs/downscoping-short-lived-credentials
"""
+import datetime
+
+from google.auth import _helpers
+from google.auth import credentials
+from google.oauth2 import sts
+
# The maximum number of access boundary rules a Credential Access Boundary can
# contain.
_MAX_ACCESS_BOUNDARY_RULES_COUNT = 10
+# The token exchange grant_type used for exchanging credentials.
+_STS_GRANT_TYPE = "urn:ietf:params:oauth:grant-type:token-exchange"
+# The token exchange requested_token_type. This is always an access_token.
+_STS_REQUESTED_TOKEN_TYPE = "urn:ietf:params:oauth:token-type:access_token"
+# The STS token URL used to exchanged a short lived access token for a downscoped one.
+_STS_TOKEN_URL = "https://sts.googleapis.com/v1/token"
+# The subject token type to use when exchanging a short lived access token for a
+# downscoped token.
+_STS_SUBJECT_TOKEN_TYPE = "urn:ietf:params:oauth:token-type:access_token"
class CredentialAccessBoundary(object):
@@ -403,3 +418,74 @@
if self.description:
json["description"] = self.description
return json
+
+
+class Credentials(credentials.CredentialsWithQuotaProject):
+ """Defines a set of Google credentials that are downscoped from an existing set
+ of Google OAuth2 credentials. This is useful to restrict the Identity and Access
+ Management (IAM) permissions that a short-lived credential can use.
+ The common pattern of usage is to have a token broker with elevated access
+ generate these downscoped credentials from higher access source credentials and
+ pass the downscoped short-lived access tokens to a token consumer via some
+ secure authenticated channel for limited access to Google Cloud Storage
+ resources.
+ """
+
+ def __init__(
+ self, source_credentials, credential_access_boundary, quota_project_id=None
+ ):
+ """Instantiates a downscoped credentials object using the provided source
+ credentials and credential access boundary rules.
+ To downscope permissions of a source credential, a Credential Access Boundary
+ that specifies which resources the new credential can access, as well as an
+ upper bound on the permissions that are available on each resource, has to be
+ defined. A downscoped credential can then be instantiated using the source
+ credential and the Credential Access Boundary.
+
+ Args:
+ source_credentials (google.auth.credentials.Credentials): The source credentials
+ to be downscoped based on the provided Credential Access Boundary rules.
+ credential_access_boundary (google.auth.downscoped.CredentialAccessBoundary):
+ The Credential Access Boundary which contains a list of access boundary
+ rules. Each rule contains information on the resource that the rule applies to,
+ the upper bound of the permissions that are available on that resource and an
+ optional condition to further restrict permissions.
+ quota_project_id (Optional[str]): The optional quota project ID.
+ Raises:
+ google.auth.exceptions.RefreshError: If the source credentials
+ return an error on token refresh.
+ google.auth.exceptions.OAuthError: If the STS token exchange
+ endpoint returned an error during downscoped token generation.
+ """
+
+ super(Credentials, self).__init__()
+ self._source_credentials = source_credentials
+ self._credential_access_boundary = credential_access_boundary
+ self._quota_project_id = quota_project_id
+ self._sts_client = sts.Client(_STS_TOKEN_URL)
+
+ @_helpers.copy_docstring(credentials.Credentials)
+ def refresh(self, request):
+ # Generate an access token from the source credentials.
+ self._source_credentials.refresh(request)
+ now = _helpers.utcnow()
+ # Exchange the access token for a downscoped access token.
+ response_data = self._sts_client.exchange_token(
+ request=request,
+ grant_type=_STS_GRANT_TYPE,
+ subject_token=self._source_credentials.token,
+ subject_token_type=_STS_SUBJECT_TOKEN_TYPE,
+ requested_token_type=_STS_REQUESTED_TOKEN_TYPE,
+ additional_options=self._credential_access_boundary.to_json(),
+ )
+ self.token = response_data.get("access_token")
+ lifetime = datetime.timedelta(seconds=response_data.get("expires_in"))
+ self.expiry = now + lifetime
+
+ @_helpers.copy_docstring(credentials.CredentialsWithQuotaProject)
+ def with_quota_project(self, quota_project_id):
+ return self.__class__(
+ self._source_credentials,
+ self._credential_access_boundary,
+ quota_project_id=quota_project_id,
+ )