Shortcuts on this page
r m x toggle line displays
j k next/prev highlighted chunk
0 (zero) top of page
1 (one) first highlighted chunk
1# Copyright 2022 Google LLC
2#
3# Licensed under the Apache License, Version 2.0 (the "License");
4# you may not use this file except in compliance with the License.
5# You may obtain a copy of the License at
6#
7# http://www.apache.org/licenses/LICENSE-2.0
8#
9# Unless required by applicable law or agreed to in writing, software
10# distributed under the License is distributed on an "AS IS" BASIS,
11# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12# See the License for the specific language governing permissions and
13# limitations under the License.
15"""External Account Authorized User Credentials.
16This module provides credentials based on OAuth 2.0 access and refresh tokens.
17These credentials usually access resources on behalf of a user (resource
18owner).
20Specifically, these are sourced using external identities via Workforce Identity Federation.
22Obtaining the initial access and refresh token can be done through the Google Cloud CLI.
24Example credential:
25{
26 "type": "external_account_authorized_user",
27 "audience": "//iam.googleapis.com/locations/global/workforcePools/$WORKFORCE_POOL_ID/providers/$PROVIDER_ID",
28 "refresh_token": "refreshToken",
29 "token_url": "https://sts.googleapis.com/v1/oauth/token",
30 "token_info_url": "https://sts.googleapis.com/v1/instrospect",
31 "client_id": "clientId",
32 "client_secret": "clientSecret"
33}
34"""
36import datetime
37import io
38import json
40from google.auth import _helpers
41from google.auth import credentials
42from google.auth import exceptions
43from google.oauth2 import sts
44from google.oauth2 import utils
46_EXTERNAL_ACCOUNT_AUTHORIZED_USER_JSON_TYPE = "external_account_authorized_user"
49class Credentials(
50 credentials.CredentialsWithQuotaProject,
51 credentials.ReadOnlyScoped,
52 credentials.CredentialsWithTokenUri,
53):
54 """Credentials for External Account Authorized Users.
56 This is used to instantiate Credentials for exchanging refresh tokens from
57 authorized users for Google access token and authorizing requests to Google
58 APIs.
60 The credentials are considered immutable. If you want to modify the
61 quota project, use `with_quota_project` and if you want to modify the token
62 uri, use `with_token_uri`.
63 """
65 def __init__(
66 self,
67 token=None,
68 expiry=None,
69 refresh_token=None,
70 audience=None,
71 client_id=None,
72 client_secret=None,
73 token_url=None,
74 token_info_url=None,
75 revoke_url=None,
76 scopes=None,
77 quota_project_id=None,
78 universe_domain=credentials.DEFAULT_UNIVERSE_DOMAIN,
79 ):
80 """Instantiates a external account authorized user credentials object.
82 Args:
83 token (str): The OAuth 2.0 access token. Can be None if refresh information
84 is provided.
85 expiry (datetime.datetime): The optional expiration datetime of the OAuth 2.0 access
86 token.
87 refresh_token (str): The optional OAuth 2.0 refresh token. If specified,
88 credentials can be refreshed.
89 audience (str): The optional STS audience which contains the resource name for the workforce
90 pool and the provider identifier in that pool.
91 client_id (str): The OAuth 2.0 client ID. Must be specified for refresh, can be left as
92 None if the token can not be refreshed.
93 client_secret (str): The OAuth 2.0 client secret. Must be specified for refresh, can be
94 left as None if the token can not be refreshed.
95 token_url (str): The optional STS token exchange endpoint for refresh. Must be specified for
96 refresh, can be left as None if the token can not be refreshed.
97 token_info_url (str): The optional STS endpoint URL for token introspection.
98 revoke_url (str): The optional STS endpoint URL for revoking tokens.
99 quota_project_id (str): The optional project ID used for quota and billing.
100 This project may be different from the project used to
101 create the credentials.
102 universe_domain (Optional[str]): The universe domain. The default value
103 is googleapis.com.
105 Returns:
106 google.auth.external_account_authorized_user.Credentials: The
107 constructed credentials.
108 """
109 super(Credentials, self).__init__()
111 self.token = token
112 self.expiry = expiry
113 self._audience = audience
114 self._refresh_token = refresh_token
115 self._token_url = token_url
116 self._token_info_url = token_info_url
117 self._client_id = client_id
118 self._client_secret = client_secret
119 self._revoke_url = revoke_url
120 self._quota_project_id = quota_project_id
121 self._scopes = scopes
122 self._universe_domain = universe_domain or credentials.DEFAULT_UNIVERSE_DOMAIN
123 self._cred_file_path = None
125 if not self.valid and not self.can_refresh:
126 raise exceptions.InvalidOperation(
127 "Token should be created with fields to make it valid (`token` and "
128 "`expiry`), or fields to allow it to refresh (`refresh_token`, "
129 "`token_url`, `client_id`, `client_secret`)."
130 )
132 self._client_auth = None
133 if self._client_id:
134 self._client_auth = utils.ClientAuthentication(
135 utils.ClientAuthType.basic, self._client_id, self._client_secret
136 )
137 self._sts_client = sts.Client(self._token_url, self._client_auth)
139 @property
140 def info(self):
141 """Generates the serializable dictionary representation of the current
142 credentials.
144 Returns:
145 Mapping: The dictionary representation of the credentials. This is the
146 reverse of the "from_info" method defined in this class. It is
147 useful for serializing the current credentials so it can deserialized
148 later.
149 """
150 config_info = self.constructor_args()
151 config_info.update(type=_EXTERNAL_ACCOUNT_AUTHORIZED_USER_JSON_TYPE)
152 if config_info["expiry"]:
153 config_info["expiry"] = config_info["expiry"].isoformat() + "Z"
155 return {key: value for key, value in config_info.items() if value is not None}
157 def constructor_args(self):
158 return {
159 "audience": self._audience,
160 "refresh_token": self._refresh_token,
161 "token_url": self._token_url,
162 "token_info_url": self._token_info_url,
163 "client_id": self._client_id,
164 "client_secret": self._client_secret,
165 "token": self.token,
166 "expiry": self.expiry,
167 "revoke_url": self._revoke_url,
168 "scopes": self._scopes,
169 "quota_project_id": self._quota_project_id,
170 "universe_domain": self._universe_domain,
171 }
173 @property
174 def scopes(self):
175 """Optional[str]: The OAuth 2.0 permission scopes."""
176 return self._scopes
178 @property
179 def requires_scopes(self):
180 """ False: OAuth 2.0 credentials have their scopes set when
181 the initial token is requested and can not be changed."""
182 return False
184 @property
185 def client_id(self):
186 """Optional[str]: The OAuth 2.0 client ID."""
187 return self._client_id
189 @property
190 def client_secret(self):
191 """Optional[str]: The OAuth 2.0 client secret."""
192 return self._client_secret
194 @property
195 def audience(self):
196 """Optional[str]: The STS audience which contains the resource name for the
197 workforce pool and the provider identifier in that pool."""
198 return self._audience
200 @property
201 def refresh_token(self):
202 """Optional[str]: The OAuth 2.0 refresh token."""
203 return self._refresh_token
205 @property
206 def token_url(self):
207 """Optional[str]: The STS token exchange endpoint for refresh."""
208 return self._token_url
210 @property
211 def token_info_url(self):
212 """Optional[str]: The STS endpoint for token info."""
213 return self._token_info_url
215 @property
216 def revoke_url(self):
217 """Optional[str]: The STS endpoint for token revocation."""
218 return self._revoke_url
220 @property
221 def is_user(self):
222 """ True: This credential always represents a user."""
223 return True
225 @property
226 def can_refresh(self):
227 return all(
228 (self._refresh_token, self._token_url, self._client_id, self._client_secret)
229 )
231 def get_project_id(self, request=None):
232 """Retrieves the project ID corresponding to the workload identity or workforce pool.
233 For workforce pool credentials, it returns the project ID corresponding to
234 the workforce_pool_user_project.
236 When not determinable, None is returned.
238 Args:
239 request (google.auth.transport.requests.Request): Request object.
240 Unused here, but passed from _default.default().
242 Return:
243 str: project ID is not determinable for this credential type so it returns None
244 """
246 return None
248 def to_json(self, strip=None):
249 """Utility function that creates a JSON representation of this
250 credential.
251 Args:
252 strip (Sequence[str]): Optional list of members to exclude from the
253 generated JSON.
254 Returns:
255 str: A JSON representation of this instance. When converted into
256 a dictionary, it can be passed to from_info()
257 to create a new instance.
258 """
259 strip = strip if strip else []
260 return json.dumps({k: v for (k, v) in self.info.items() if k not in strip})
262 def refresh(self, request):
263 """Refreshes the access token.
265 Args:
266 request (google.auth.transport.Request): The object used to make
267 HTTP requests.
269 Raises:
270 google.auth.exceptions.RefreshError: If the credentials could
271 not be refreshed.
272 """
273 if not self.can_refresh:
274 raise exceptions.RefreshError(
275 "The credentials do not contain the necessary fields need to "
276 "refresh the access token. You must specify refresh_token, "
277 "token_url, client_id, and client_secret."
278 )
280 now = _helpers.utcnow()
281 response_data = self._make_sts_request(request)
283 self.token = response_data.get("access_token")
285 lifetime = datetime.timedelta(seconds=response_data.get("expires_in"))
286 self.expiry = now + lifetime
288 if "refresh_token" in response_data:
289 self._refresh_token = response_data["refresh_token"]
291 def _make_sts_request(self, request):
292 return self._sts_client.refresh_token(request, self._refresh_token)
294 @_helpers.copy_docstring(credentials.Credentials)
295 def get_cred_info(self):
296 if self._cred_file_path:
297 return {
298 "credential_source": self._cred_file_path,
299 "credential_type": "external account authorized user credentials",
300 }
301 return None
303 def _make_copy(self):
304 kwargs = self.constructor_args()
305 cred = self.__class__(**kwargs)
306 cred._cred_file_path = self._cred_file_path
307 return cred
309 @_helpers.copy_docstring(credentials.CredentialsWithQuotaProject)
310 def with_quota_project(self, quota_project_id):
311 cred = self._make_copy()
312 cred._quota_project_id = quota_project_id
313 return cred
315 @_helpers.copy_docstring(credentials.CredentialsWithTokenUri)
316 def with_token_uri(self, token_uri):
317 cred = self._make_copy()
318 cred._token_url = token_uri
319 return cred
321 @_helpers.copy_docstring(credentials.CredentialsWithUniverseDomain)
322 def with_universe_domain(self, universe_domain):
323 cred = self._make_copy()
324 cred._universe_domain = universe_domain
325 return cred
327 @classmethod
328 def from_info(cls, info, **kwargs):
329 """Creates a Credentials instance from parsed external account info.
331 Args:
332 info (Mapping[str, str]): The external account info in Google
333 format.
334 kwargs: Additional arguments to pass to the constructor.
336 Returns:
337 google.auth.external_account_authorized_user.Credentials: The
338 constructed credentials.
340 Raises:
341 ValueError: For invalid parameters.
342 """
343 expiry = info.get("expiry")
344 if expiry:
345 expiry = datetime.datetime.strptime(
346 expiry.rstrip("Z").split(".")[0], "%Y-%m-%dT%H:%M:%S"
347 )
348 return cls(
349 audience=info.get("audience"),
350 refresh_token=info.get("refresh_token"),
351 token_url=info.get("token_url"),
352 token_info_url=info.get("token_info_url"),
353 client_id=info.get("client_id"),
354 client_secret=info.get("client_secret"),
355 token=info.get("token"),
356 expiry=expiry,
357 revoke_url=info.get("revoke_url"),
358 quota_project_id=info.get("quota_project_id"),
359 scopes=info.get("scopes"),
360 universe_domain=info.get(
361 "universe_domain", credentials.DEFAULT_UNIVERSE_DOMAIN
362 ),
363 **kwargs
364 )
366 @classmethod
367 def from_file(cls, filename, **kwargs):
368 """Creates a Credentials instance from an external account json file.
370 Args:
371 filename (str): The path to the external account json file.
372 kwargs: Additional arguments to pass to the constructor.
374 Returns:
375 google.auth.external_account_authorized_user.Credentials: The
376 constructed credentials.
377 """
378 with io.open(filename, "r", encoding="utf-8") as json_file:
379 data = json.load(json_file)
380 return cls.from_info(data, **kwargs)