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 2016 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"""Service Accounts: JSON Web Token (JWT) Profile for OAuth 2.0
17This module implements the JWT Profile for OAuth 2.0 Authorization Grants
18as defined by `RFC 7523`_ with particular support for how this RFC is
19implemented in Google's infrastructure. Google refers to these credentials
20as *Service Accounts*.
22Service accounts are used for server-to-server communication, such as
23interactions between a web application server and a Google service. The
24service account belongs to your application instead of to an individual end
25user. In contrast to other OAuth 2.0 profiles, no users are involved and your
26application "acts" as the service account.
28Typically an application uses a service account when the application uses
29Google APIs to work with its own data rather than a user's data. For example,
30an application that uses Google Cloud Datastore for data persistence would use
31a service account to authenticate its calls to the Google Cloud Datastore API.
32However, an application that needs to access a user's Drive documents would
33use the normal OAuth 2.0 profile.
35Additionally, Google Apps domain administrators can grant service accounts
36`domain-wide delegation`_ authority to access user data on behalf of users in
37the domain.
39This profile uses a JWT to acquire an OAuth 2.0 access token. The JWT is used
40in place of the usual authorization token returned during the standard
41OAuth 2.0 Authorization Code grant. The JWT is only used for this purpose, as
42the acquired access token is used as the bearer token when making requests
43using these credentials.
45This profile differs from normal OAuth 2.0 profile because no user consent
46step is required. The use of the private key allows this profile to assert
47identity directly.
49This profile also differs from the :mod:`google.auth.jwt` authentication
50because the JWT credentials use the JWT directly as the bearer token. This
51profile instead only uses the JWT to obtain an OAuth 2.0 access token. The
52obtained OAuth 2.0 access token is used as the bearer token.
54Domain-wide delegation
55----------------------
57Domain-wide delegation allows a service account to access user data on
58behalf of any user in a Google Apps domain without consent from the user.
59For example, an application that uses the Google Calendar API to add events to
60the calendars of all users in a Google Apps domain would use a service account
61to access the Google Calendar API on behalf of users.
63The Google Apps administrator must explicitly authorize the service account to
64do this. This authorization step is referred to as "delegating domain-wide
65authority" to a service account.
67You can use domain-wise delegation by creating a set of credentials with a
68specific subject using :meth:`~Credentials.with_subject`.
70.. _RFC 7523: https://tools.ietf.org/html/rfc7523
71"""
73import copy
74import datetime
75import logging
76from typing import TYPE_CHECKING, Optional
78from google.auth import (
79 _helpers,
80 _regional_access_boundary_utils,
81 _service_account_info,
82 credentials,
83 exceptions,
84 iam,
85 jwt,
86 metrics,
87)
88from google.oauth2 import _client
90if TYPE_CHECKING: # pragma: NO COVER
91 import google.auth.transport
93_LOGGER = logging.getLogger(__name__)
95_DEFAULT_TOKEN_LIFETIME_SECS = 3600 # 1 hour in seconds
96_GOOGLE_OAUTH2_TOKEN_ENDPOINT = "https://oauth2.googleapis.com/token"
99class Credentials(
100 credentials.Signing,
101 credentials.Scoped,
102 credentials.CredentialsWithQuotaProject,
103 credentials.CredentialsWithTokenUri,
104 credentials.CredentialsWithRegionalAccessBoundary,
105):
106 """Service account credentials
108 Usually, you'll create these credentials with one of the helper
109 constructors. To create credentials using a Google service account
110 private key JSON file::
112 credentials = service_account.Credentials.from_service_account_file(
113 'service-account.json')
115 Or if you already have the service account file loaded::
117 service_account_info = json.load(open('service_account.json'))
118 credentials = service_account.Credentials.from_service_account_info(
119 service_account_info)
121 Both helper methods pass on arguments to the constructor, so you can
122 specify additional scopes and a subject if necessary::
124 credentials = service_account.Credentials.from_service_account_file(
125 'service-account.json',
126 scopes=['email'],
127 subject='user@example.com')
129 The credentials are considered immutable. If you want to modify the scopes
130 or the subject used for delegation, use :meth:`with_scopes` or
131 :meth:`with_subject`::
133 scoped_credentials = credentials.with_scopes(['email'])
134 delegated_credentials = credentials.with_subject(subject)
136 To add a quota project, use :meth:`with_quota_project`::
138 credentials = credentials.with_quota_project('myproject-123')
139 """
141 def __init__(
142 self,
143 signer,
144 service_account_email,
145 token_uri,
146 scopes=None,
147 default_scopes=None,
148 subject=None,
149 project_id=None,
150 quota_project_id=None,
151 additional_claims=None,
152 always_use_jwt_access=False,
153 universe_domain=credentials.DEFAULT_UNIVERSE_DOMAIN,
154 trust_boundary=None,
155 ):
156 """
157 Args:
158 signer (google.auth.crypt.Signer): The signer used to sign JWTs.
159 service_account_email (str): The service account's email.
160 scopes (Sequence[str]): User-defined scopes to request during the
161 authorization grant.
162 default_scopes (Sequence[str]): Default scopes passed by a
163 Google client library. Use 'scopes' for user-defined scopes.
164 token_uri (str): The OAuth 2.0 Token URI.
165 subject (str): For domain-wide delegation, the email address of the
166 user to for which to request delegated access.
167 project_id (str): Project ID associated with the service account
168 credential.
169 quota_project_id (Optional[str]): The project ID used for quota and
170 billing.
171 additional_claims (Mapping[str, str]): Any additional claims for
172 the JWT assertion used in the authorization grant.
173 always_use_jwt_access (Optional[bool]): Whether self signed JWT should
174 be always used.
175 universe_domain (str): The universe domain. The default
176 universe domain is googleapis.com. For default value self
177 signed jwt is used for token refresh.
178 trust_boundary (Mapping[str,str]): A credential trust boundary.
180 .. note:: Typically one of the helper constructors
181 :meth:`from_service_account_file` or
182 :meth:`from_service_account_info` are used instead of calling the
183 constructor directly.
184 """
185 super(Credentials, self).__init__()
187 self._cred_file_path = None
188 self._scopes = scopes
189 self._default_scopes = default_scopes
190 self._signer = signer
191 self._service_account_email = service_account_email
192 self._subject = subject
193 self._project_id = project_id
194 self._quota_project_id = quota_project_id
195 self._token_uri = token_uri
196 self._always_use_jwt_access = always_use_jwt_access
197 self._universe_domain = universe_domain or credentials.DEFAULT_UNIVERSE_DOMAIN
199 if universe_domain != credentials.DEFAULT_UNIVERSE_DOMAIN:
200 self._always_use_jwt_access = True
202 self._jwt_credentials = None
204 if additional_claims is not None:
205 self._additional_claims = additional_claims
206 else:
207 self._additional_claims = {}
209 self._trust_boundary = trust_boundary
211 @classmethod
212 def _from_signer_and_info(cls, signer, info, **kwargs):
213 """Creates a Credentials instance from a signer and service account
214 info.
216 Args:
217 signer (google.auth.crypt.Signer): The signer used to sign JWTs.
218 info (Mapping[str, str]): The service account info.
219 kwargs: Additional arguments to pass to the constructor.
221 Returns:
222 google.auth.jwt.Credentials: The constructed credentials.
224 Raises:
225 ValueError: If the info is not in the expected format.
226 """
227 return cls(
228 signer,
229 service_account_email=info["client_email"],
230 token_uri=info["token_uri"],
231 project_id=info.get("project_id"),
232 universe_domain=info.get(
233 "universe_domain", credentials.DEFAULT_UNIVERSE_DOMAIN
234 ),
235 trust_boundary=info.get("trust_boundary"),
236 **kwargs,
237 )
239 @classmethod
240 def from_service_account_info(cls, info, **kwargs):
241 """Creates a Credentials instance from parsed service account info.
243 Args:
244 info (Mapping[str, str]): The service account info in Google
245 format.
246 kwargs: Additional arguments to pass to the constructor.
248 Returns:
249 google.auth.service_account.Credentials: The constructed
250 credentials.
252 Raises:
253 ValueError: If the info is not in the expected format.
254 """
255 signer = _service_account_info.from_dict(
256 info, require=["client_email", "token_uri"]
257 )
258 return cls._from_signer_and_info(signer, info, **kwargs)
260 @classmethod
261 def from_service_account_file(cls, filename, **kwargs):
262 """Creates a Credentials instance from a service account json file.
264 Args:
265 filename (str): The path to the service account json file.
266 kwargs: Additional arguments to pass to the constructor.
268 Returns:
269 google.auth.service_account.Credentials: The constructed
270 credentials.
271 """
272 info, signer = _service_account_info.from_filename(
273 filename, require=["client_email", "token_uri"]
274 )
275 return cls._from_signer_and_info(signer, info, **kwargs)
277 @property
278 def service_account_email(self):
279 """The service account email."""
280 return self._service_account_email
282 @property
283 def project_id(self):
284 """Project ID associated with this credential."""
285 return self._project_id
287 @property
288 def requires_scopes(self):
289 """Checks if the credentials requires scopes.
291 Returns:
292 bool: True if there are no scopes set otherwise False.
293 """
294 return True if not self._scopes else False
296 def _make_copy(self):
297 cred = self.__class__(
298 self._signer,
299 service_account_email=self._service_account_email,
300 scopes=copy.copy(self._scopes),
301 default_scopes=copy.copy(self._default_scopes),
302 token_uri=self._token_uri,
303 subject=self._subject,
304 project_id=self._project_id,
305 quota_project_id=self._quota_project_id,
306 additional_claims=self._additional_claims.copy(),
307 always_use_jwt_access=self._always_use_jwt_access,
308 universe_domain=self._universe_domain,
309 trust_boundary=self._trust_boundary,
310 )
311 cred._cred_file_path = self._cred_file_path
312 self._copy_regional_access_boundary_manager(cred)
313 return cred
315 @_helpers.copy_docstring(credentials.Scoped)
316 def with_scopes(self, scopes, default_scopes=None):
317 cred = self._make_copy()
318 cred._scopes = scopes
319 cred._default_scopes = default_scopes
320 return cred
322 def with_always_use_jwt_access(self, always_use_jwt_access):
323 """Create a copy of these credentials with the specified always_use_jwt_access value.
325 Args:
326 always_use_jwt_access (bool): Whether always use self signed JWT or not.
328 Returns:
329 google.auth.service_account.Credentials: A new credentials
330 instance.
331 Raises:
332 google.auth.exceptions.InvalidValue: If the universe domain is not
333 default and always_use_jwt_access is False.
334 """
335 cred = self._make_copy()
336 if (
337 cred._universe_domain != credentials.DEFAULT_UNIVERSE_DOMAIN
338 and not always_use_jwt_access
339 ):
340 raise exceptions.InvalidValue(
341 "always_use_jwt_access should be True for non-default universe domain"
342 )
343 cred._always_use_jwt_access = always_use_jwt_access
344 return cred
346 @_helpers.copy_docstring(credentials.CredentialsWithUniverseDomain)
347 def with_universe_domain(self, universe_domain):
348 cred = self._make_copy()
349 cred._universe_domain = universe_domain
350 if universe_domain != credentials.DEFAULT_UNIVERSE_DOMAIN:
351 cred._always_use_jwt_access = True
352 return cred
354 def with_subject(self, subject):
355 """Create a copy of these credentials with the specified subject.
357 Args:
358 subject (str): The subject claim.
360 Returns:
361 google.auth.service_account.Credentials: A new credentials
362 instance.
363 """
364 cred = self._make_copy()
365 cred._subject = subject
366 return cred
368 def with_claims(self, additional_claims):
369 """Returns a copy of these credentials with modified claims.
371 Args:
372 additional_claims (Mapping[str, str]): Any additional claims for
373 the JWT payload. This will be merged with the current
374 additional claims.
376 Returns:
377 google.auth.service_account.Credentials: A new credentials
378 instance.
379 """
380 new_additional_claims = copy.deepcopy(self._additional_claims)
381 new_additional_claims.update(additional_claims or {})
382 cred = self._make_copy()
383 cred._additional_claims = new_additional_claims
384 return cred
386 @_helpers.copy_docstring(credentials.CredentialsWithQuotaProject)
387 def with_quota_project(self, quota_project_id):
388 cred = self._make_copy()
389 cred._quota_project_id = quota_project_id
390 return cred
392 @_helpers.copy_docstring(credentials.CredentialsWithTokenUri)
393 def with_token_uri(self, token_uri):
394 cred = self._make_copy()
395 cred._token_uri = token_uri
396 return cred
398 def _make_authorization_grant_assertion(self):
399 """Create the OAuth 2.0 assertion.
401 This assertion is used during the OAuth 2.0 grant to acquire an
402 access token.
404 Returns:
405 bytes: The authorization grant assertion.
406 """
407 now = _helpers.utcnow()
408 lifetime = datetime.timedelta(seconds=_DEFAULT_TOKEN_LIFETIME_SECS)
409 expiry = now + lifetime
411 payload = {
412 "iat": _helpers.datetime_to_secs(now),
413 "exp": _helpers.datetime_to_secs(expiry),
414 # The issuer must be the service account email.
415 "iss": self._service_account_email,
416 # The audience must be the auth token endpoint's URI
417 "aud": _GOOGLE_OAUTH2_TOKEN_ENDPOINT,
418 "scope": _helpers.scopes_to_string(self._scopes or ()),
419 }
421 payload.update(self._additional_claims)
423 # The subject can be a user email for domain-wide delegation.
424 if self._subject:
425 payload.setdefault("sub", self._subject)
427 token = jwt.encode(self._signer, payload)
429 return token
431 def _use_self_signed_jwt(self):
432 # Since domain wide delegation doesn't work with self signed JWT. If
433 # subject exists, then we should not use self signed JWT.
434 return self._subject is None and self._jwt_credentials is not None
436 def _metric_header_for_usage(self):
437 if self._use_self_signed_jwt():
438 return metrics.CRED_TYPE_SA_JWT
439 return metrics.CRED_TYPE_SA_ASSERTION
441 @_helpers.copy_docstring(credentials.CredentialsWithRegionalAccessBoundary)
442 def _perform_refresh_token(self, request):
443 if self._always_use_jwt_access and not self._jwt_credentials:
444 # If self signed jwt should be used but jwt credential is not
445 # created, try to create one with scopes
446 self._create_self_signed_jwt(None)
448 if (
449 self._universe_domain != credentials.DEFAULT_UNIVERSE_DOMAIN
450 and self._subject
451 ):
452 raise exceptions.RefreshError(
453 "domain wide delegation is not supported for non-default universe domain"
454 )
456 if self._use_self_signed_jwt():
457 self._jwt_credentials.refresh(request)
458 self.token = self._jwt_credentials.token.decode()
459 self.expiry = self._jwt_credentials.expiry
460 else:
461 assertion = self._make_authorization_grant_assertion()
462 access_token, expiry, _ = _client.jwt_grant(
463 request, self._token_uri, assertion
464 )
465 self.token = access_token
466 self.expiry = expiry
468 def _create_self_signed_jwt(self, audience):
469 """Create a self-signed JWT from the credentials if requirements are met.
471 Args:
472 audience (str): The service URL. ``https://[API_ENDPOINT]/``
473 """
474 # https://google.aip.dev/auth/4111
475 if self._always_use_jwt_access:
476 if self._scopes:
477 additional_claims = {"scope": " ".join(self._scopes)}
478 if (
479 self._jwt_credentials is None
480 or self._jwt_credentials.additional_claims != additional_claims
481 ):
482 self._jwt_credentials = jwt.Credentials.from_signing_credentials(
483 self, None, additional_claims=additional_claims
484 )
485 elif audience:
486 if (
487 self._jwt_credentials is None
488 or self._jwt_credentials._audience != audience
489 ):
490 self._jwt_credentials = jwt.Credentials.from_signing_credentials(
491 self, audience
492 )
493 elif self._default_scopes:
494 additional_claims = {"scope": " ".join(self._default_scopes)}
495 if (
496 self._jwt_credentials is None
497 or additional_claims != self._jwt_credentials.additional_claims
498 ):
499 self._jwt_credentials = jwt.Credentials.from_signing_credentials(
500 self, None, additional_claims=additional_claims
501 )
502 elif not self._scopes and audience:
503 self._jwt_credentials = jwt.Credentials.from_signing_credentials(
504 self, audience
505 )
507 def _build_regional_access_boundary_lookup_url(
508 self,
509 request: "Optional[google.auth.transport.Request]" = None, # noqa: F821
510 ):
511 """Builds and returns the URL for the Regional Access Boundary lookup API.
513 This method constructs the specific URL for the IAM Credentials API's
514 `allowedLocations` endpoint, using the credential's universe domain
515 and service account email.
517 Returns:
518 Optional[str]: The URL for the Regional Access Boundary lookup endpoint, or None
519 if the service account email is missing. Returns None if the subject is populated.
520 """
521 if self._subject:
522 # RAB does not apply to Workspace User Accounts via Domain-wide Delegation.
523 return None
525 if not self.service_account_email:
526 _LOGGER.error(
527 "Service account email is required to build the Regional Access Boundary lookup URL for service account credentials."
528 )
529 return None
530 return _regional_access_boundary_utils.get_service_account_rab_endpoint(
531 self._service_account_email
532 )
534 @_helpers.copy_docstring(credentials.Signing)
535 def sign_bytes(self, message):
536 return self._signer.sign(message)
538 @property # type: ignore
539 @_helpers.copy_docstring(credentials.Signing)
540 def signer(self):
541 return self._signer
543 @property # type: ignore
544 @_helpers.copy_docstring(credentials.Signing)
545 def signer_email(self):
546 return self._service_account_email
548 @_helpers.copy_docstring(credentials.Credentials)
549 def get_cred_info(self):
550 if self._cred_file_path:
551 return {
552 "credential_source": self._cred_file_path,
553 "credential_type": "service account credentials",
554 "principal": self.service_account_email,
555 }
556 return None
559class IDTokenCredentials(
560 credentials.Signing,
561 credentials.CredentialsWithQuotaProject,
562 credentials.CredentialsWithTokenUri,
563):
564 """Open ID Connect ID Token-based service account credentials.
566 These credentials are largely similar to :class:`.Credentials`, but instead
567 of using an OAuth 2.0 Access Token as the bearer token, they use an Open
568 ID Connect ID Token as the bearer token. These credentials are useful when
569 communicating to services that require ID Tokens and can not accept access
570 tokens.
572 Usually, you'll create these credentials with one of the helper
573 constructors. To create credentials using a Google service account
574 private key JSON file::
576 credentials = (
577 service_account.IDTokenCredentials.from_service_account_file(
578 'service-account.json'))
581 Or if you already have the service account file loaded::
583 service_account_info = json.load(open('service_account.json'))
584 credentials = (
585 service_account.IDTokenCredentials.from_service_account_info(
586 service_account_info))
589 Both helper methods pass on arguments to the constructor, so you can
590 specify additional scopes and a subject if necessary::
592 credentials = (
593 service_account.IDTokenCredentials.from_service_account_file(
594 'service-account.json',
595 scopes=['email'],
596 subject='user@example.com'))
599 The credentials are considered immutable. If you want to modify the scopes
600 or the subject used for delegation, use :meth:`with_scopes` or
601 :meth:`with_subject`::
603 scoped_credentials = credentials.with_scopes(['email'])
604 delegated_credentials = credentials.with_subject(subject)
606 """
608 def __init__(
609 self,
610 signer,
611 service_account_email,
612 token_uri,
613 target_audience,
614 additional_claims=None,
615 quota_project_id=None,
616 universe_domain=credentials.DEFAULT_UNIVERSE_DOMAIN,
617 ):
618 """
619 Args:
620 signer (google.auth.crypt.Signer): The signer used to sign JWTs.
621 service_account_email (str): The service account's email.
622 token_uri (str): The OAuth 2.0 Token URI.
623 target_audience (str): The intended audience for these credentials,
624 used when requesting the ID Token. The ID Token's ``aud`` claim
625 will be set to this string.
626 additional_claims (Mapping[str, str]): Any additional claims for
627 the JWT assertion used in the authorization grant.
628 quota_project_id (Optional[str]): The project ID used for quota and billing.
629 universe_domain (str): The universe domain. The default
630 universe domain is googleapis.com. For default value IAM ID
631 token endponint is used for token refresh. Note that
632 iam.serviceAccountTokenCreator role is required to use the IAM
633 endpoint.
635 .. note:: Typically one of the helper constructors
636 :meth:`from_service_account_file` or
637 :meth:`from_service_account_info` are used instead of calling the
638 constructor directly.
639 """
640 super(IDTokenCredentials, self).__init__()
641 self._signer = signer
642 self._service_account_email = service_account_email
643 self._token_uri = token_uri
644 self._target_audience = target_audience
645 self._quota_project_id = quota_project_id
646 self._use_iam_endpoint = False
648 if not universe_domain:
649 self._universe_domain = credentials.DEFAULT_UNIVERSE_DOMAIN
650 else:
651 self._universe_domain = universe_domain
652 self._iam_id_token_endpoint = iam._IAM_IDTOKEN_ENDPOINT.replace(
653 "googleapis.com", self._universe_domain
654 )
656 if self._universe_domain != credentials.DEFAULT_UNIVERSE_DOMAIN:
657 self._use_iam_endpoint = True
659 if additional_claims is not None:
660 self._additional_claims = additional_claims
661 else:
662 self._additional_claims = {}
664 @classmethod
665 def _from_signer_and_info(cls, signer, info, **kwargs):
666 """Creates a credentials instance from a signer and service account
667 info.
669 Args:
670 signer (google.auth.crypt.Signer): The signer used to sign JWTs.
671 info (Mapping[str, str]): The service account info.
672 kwargs: Additional arguments to pass to the constructor.
674 Returns:
675 google.auth.jwt.IDTokenCredentials: The constructed credentials.
677 Raises:
678 ValueError: If the info is not in the expected format.
679 """
680 kwargs.setdefault("service_account_email", info["client_email"])
681 kwargs.setdefault("token_uri", info["token_uri"])
682 if "universe_domain" in info:
683 kwargs["universe_domain"] = info["universe_domain"]
684 return cls(signer, **kwargs)
686 @classmethod
687 def from_service_account_info(cls, info, **kwargs):
688 """Creates a credentials instance from parsed service account info.
690 Args:
691 info (Mapping[str, str]): The service account info in Google
692 format.
693 kwargs: Additional arguments to pass to the constructor.
695 Returns:
696 google.auth.service_account.IDTokenCredentials: The constructed
697 credentials.
699 Raises:
700 ValueError: If the info is not in the expected format.
701 """
702 signer = _service_account_info.from_dict(
703 info, require=["client_email", "token_uri"]
704 )
705 return cls._from_signer_and_info(signer, info, **kwargs)
707 @classmethod
708 def from_service_account_file(cls, filename, **kwargs):
709 """Creates a credentials instance from a service account json file.
711 Args:
712 filename (str): The path to the service account json file.
713 kwargs: Additional arguments to pass to the constructor.
715 Returns:
716 google.auth.service_account.IDTokenCredentials: The constructed
717 credentials.
718 """
719 info, signer = _service_account_info.from_filename(
720 filename, require=["client_email", "token_uri"]
721 )
722 return cls._from_signer_and_info(signer, info, **kwargs)
724 def _make_copy(self):
725 cred = self.__class__(
726 self._signer,
727 service_account_email=self._service_account_email,
728 token_uri=self._token_uri,
729 target_audience=self._target_audience,
730 additional_claims=self._additional_claims.copy(),
731 quota_project_id=self.quota_project_id,
732 universe_domain=self._universe_domain,
733 )
734 # _use_iam_endpoint is not exposed in the constructor
735 cred._use_iam_endpoint = self._use_iam_endpoint
736 return cred
738 def with_target_audience(self, target_audience):
739 """Create a copy of these credentials with the specified target
740 audience.
742 Args:
743 target_audience (str): The intended audience for these credentials,
744 used when requesting the ID Token.
746 Returns:
747 google.auth.service_account.IDTokenCredentials: A new credentials
748 instance.
749 """
750 cred = self._make_copy()
751 cred._target_audience = target_audience
752 return cred
754 def _with_use_iam_endpoint(self, use_iam_endpoint):
755 """Create a copy of these credentials with the use_iam_endpoint value.
757 Args:
758 use_iam_endpoint (bool): If True, IAM generateIdToken endpoint will
759 be used instead of the token_uri. Note that
760 iam.serviceAccountTokenCreator role is required to use the IAM
761 endpoint. The default value is False. This feature is currently
762 experimental and subject to change without notice.
764 Returns:
765 google.auth.service_account.IDTokenCredentials: A new credentials
766 instance.
767 Raises:
768 google.auth.exceptions.InvalidValue: If the universe domain is not
769 default and use_iam_endpoint is False.
770 """
771 cred = self._make_copy()
772 if (
773 cred._universe_domain != credentials.DEFAULT_UNIVERSE_DOMAIN
774 and not use_iam_endpoint
775 ):
776 raise exceptions.InvalidValue(
777 "use_iam_endpoint should be True for non-default universe domain"
778 )
779 cred._use_iam_endpoint = use_iam_endpoint
780 return cred
782 @_helpers.copy_docstring(credentials.CredentialsWithQuotaProject)
783 def with_quota_project(self, quota_project_id):
784 cred = self._make_copy()
785 cred._quota_project_id = quota_project_id
786 return cred
788 @_helpers.copy_docstring(credentials.CredentialsWithTokenUri)
789 def with_token_uri(self, token_uri):
790 cred = self._make_copy()
791 cred._token_uri = token_uri
792 return cred
794 def _make_authorization_grant_assertion(self):
795 """Create the OAuth 2.0 assertion.
797 This assertion is used during the OAuth 2.0 grant to acquire an
798 ID token.
800 Returns:
801 bytes: The authorization grant assertion.
802 """
803 now = _helpers.utcnow()
804 lifetime = datetime.timedelta(seconds=_DEFAULT_TOKEN_LIFETIME_SECS)
805 expiry = now + lifetime
807 payload = {
808 "iat": _helpers.datetime_to_secs(now),
809 "exp": _helpers.datetime_to_secs(expiry),
810 # The issuer must be the service account email.
811 "iss": self.service_account_email,
812 # The audience must be the auth token endpoint's URI
813 "aud": _GOOGLE_OAUTH2_TOKEN_ENDPOINT,
814 # The target audience specifies which service the ID token is
815 # intended for.
816 "target_audience": self._target_audience,
817 }
819 payload.update(self._additional_claims)
821 token = jwt.encode(self._signer, payload)
823 return token
825 def _refresh_with_iam_endpoint(self, request):
826 """Use IAM generateIdToken endpoint to obtain an ID token.
828 It works as follows:
830 1. First we create a self signed jwt with
831 https://www.googleapis.com/auth/iam being the scope.
833 2. Next we use the self signed jwt as the access token, and make a POST
834 request to IAM generateIdToken endpoint. The request body is:
835 {
836 "audience": self._target_audience,
837 "includeEmail": "true",
838 "useEmailAzp": "true",
839 }
841 If the request is succesfully, it will return {"token":"the ID token"},
842 and we can extract the ID token and compute its expiry.
843 """
844 jwt_credentials = jwt.Credentials.from_signing_credentials(
845 self,
846 None,
847 additional_claims={"scope": "https://www.googleapis.com/auth/iam"},
848 )
849 jwt_credentials.refresh(request)
851 self.token, self.expiry = _client.call_iam_generate_id_token_endpoint(
852 request,
853 self._iam_id_token_endpoint,
854 self.signer_email,
855 self._target_audience,
856 jwt_credentials.token.decode(),
857 self._universe_domain,
858 )
860 @_helpers.copy_docstring(credentials.Credentials)
861 def refresh(self, request):
862 if self._use_iam_endpoint:
863 self._refresh_with_iam_endpoint(request)
864 else:
865 assertion = self._make_authorization_grant_assertion()
866 access_token, expiry, _ = _client.id_token_jwt_grant(
867 request, self._token_uri, assertion
868 )
869 self.token = access_token
870 self.expiry = expiry
872 @property
873 def service_account_email(self):
874 """The service account email."""
875 return self._service_account_email
877 @_helpers.copy_docstring(credentials.Signing)
878 def sign_bytes(self, message):
879 return self._signer.sign(message)
881 @property # type: ignore
882 @_helpers.copy_docstring(credentials.Signing)
883 def signer(self):
884 return self._signer
886 @property # type: ignore
887 @_helpers.copy_docstring(credentials.Signing)
888 def signer_email(self):
889 return self._service_account_email