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 Optional, TYPE_CHECKING
79from google.auth import _helpers
80from google.auth import _regional_access_boundary_utils
81from google.auth import _service_account_info
82from google.auth import credentials
83from google.auth import exceptions
84from google.auth import iam
85from google.auth import jwt
86from google.auth import metrics
87from google.oauth2 import _client
89if TYPE_CHECKING: # pragma: NO COVER
90 import google.auth.transport
92_LOGGER = logging.getLogger(__name__)
94_DEFAULT_TOKEN_LIFETIME_SECS = 3600 # 1 hour in seconds
95_GOOGLE_OAUTH2_TOKEN_ENDPOINT = "https://oauth2.googleapis.com/token"
98class Credentials(
99 credentials.Signing,
100 credentials.Scoped,
101 credentials.CredentialsWithQuotaProject,
102 credentials.CredentialsWithTokenUri,
103 credentials.CredentialsWithRegionalAccessBoundary,
104):
105 """Service account credentials
107 Usually, you'll create these credentials with one of the helper
108 constructors. To create credentials using a Google service account
109 private key JSON file::
111 credentials = service_account.Credentials.from_service_account_file(
112 'service-account.json')
114 Or if you already have the service account file loaded::
116 service_account_info = json.load(open('service_account.json'))
117 credentials = service_account.Credentials.from_service_account_info(
118 service_account_info)
120 Both helper methods pass on arguments to the constructor, so you can
121 specify additional scopes and a subject if necessary::
123 credentials = service_account.Credentials.from_service_account_file(
124 'service-account.json',
125 scopes=['email'],
126 subject='user@example.com')
128 The credentials are considered immutable. If you want to modify the scopes
129 or the subject used for delegation, use :meth:`with_scopes` or
130 :meth:`with_subject`::
132 scoped_credentials = credentials.with_scopes(['email'])
133 delegated_credentials = credentials.with_subject(subject)
135 To add a quota project, use :meth:`with_quota_project`::
137 credentials = credentials.with_quota_project('myproject-123')
138 """
140 def __init__(
141 self,
142 signer,
143 service_account_email,
144 token_uri,
145 scopes=None,
146 default_scopes=None,
147 subject=None,
148 project_id=None,
149 quota_project_id=None,
150 additional_claims=None,
151 always_use_jwt_access=False,
152 universe_domain=credentials.DEFAULT_UNIVERSE_DOMAIN,
153 trust_boundary=None,
154 ):
155 """
156 Args:
157 signer (google.auth.crypt.Signer): The signer used to sign JWTs.
158 service_account_email (str): The service account's email.
159 scopes (Sequence[str]): User-defined scopes to request during the
160 authorization grant.
161 default_scopes (Sequence[str]): Default scopes passed by a
162 Google client library. Use 'scopes' for user-defined scopes.
163 token_uri (str): The OAuth 2.0 Token URI.
164 subject (str): For domain-wide delegation, the email address of the
165 user to for which to request delegated access.
166 project_id (str): Project ID associated with the service account
167 credential.
168 quota_project_id (Optional[str]): The project ID used for quota and
169 billing.
170 additional_claims (Mapping[str, str]): Any additional claims for
171 the JWT assertion used in the authorization grant.
172 always_use_jwt_access (Optional[bool]): Whether self signed JWT should
173 be always used.
174 universe_domain (str): The universe domain. The default
175 universe domain is googleapis.com. For default value self
176 signed jwt is used for token refresh.
177 trust_boundary (Mapping[str,str]): A credential trust boundary.
179 .. note:: Typically one of the helper constructors
180 :meth:`from_service_account_file` or
181 :meth:`from_service_account_info` are used instead of calling the
182 constructor directly.
183 """
184 super(Credentials, self).__init__()
186 self._cred_file_path = None
187 self._scopes = scopes
188 self._default_scopes = default_scopes
189 self._signer = signer
190 self._service_account_email = service_account_email
191 self._subject = subject
192 self._project_id = project_id
193 self._quota_project_id = quota_project_id
194 self._token_uri = token_uri
195 self._always_use_jwt_access = always_use_jwt_access
196 self._universe_domain = universe_domain or credentials.DEFAULT_UNIVERSE_DOMAIN
198 if universe_domain != credentials.DEFAULT_UNIVERSE_DOMAIN:
199 self._always_use_jwt_access = True
201 self._jwt_credentials = None
203 if additional_claims is not None:
204 self._additional_claims = additional_claims
205 else:
206 self._additional_claims = {}
208 self._trust_boundary = trust_boundary
210 @classmethod
211 def _from_signer_and_info(cls, signer, info, **kwargs):
212 """Creates a Credentials instance from a signer and service account
213 info.
215 Args:
216 signer (google.auth.crypt.Signer): The signer used to sign JWTs.
217 info (Mapping[str, str]): The service account info.
218 kwargs: Additional arguments to pass to the constructor.
220 Returns:
221 google.auth.jwt.Credentials: The constructed credentials.
223 Raises:
224 ValueError: If the info is not in the expected format.
225 """
226 return cls(
227 signer,
228 service_account_email=info["client_email"],
229 token_uri=info["token_uri"],
230 project_id=info.get("project_id"),
231 universe_domain=info.get(
232 "universe_domain", credentials.DEFAULT_UNIVERSE_DOMAIN
233 ),
234 trust_boundary=info.get("trust_boundary"),
235 **kwargs,
236 )
238 @classmethod
239 def from_service_account_info(cls, info, **kwargs):
240 """Creates a Credentials instance from parsed service account info.
242 Args:
243 info (Mapping[str, str]): The service account info in Google
244 format.
245 kwargs: Additional arguments to pass to the constructor.
247 Returns:
248 google.auth.service_account.Credentials: The constructed
249 credentials.
251 Raises:
252 ValueError: If the info is not in the expected format.
253 """
254 signer = _service_account_info.from_dict(
255 info, require=["client_email", "token_uri"]
256 )
257 return cls._from_signer_and_info(signer, info, **kwargs)
259 @classmethod
260 def from_service_account_file(cls, filename, **kwargs):
261 """Creates a Credentials instance from a service account json file.
263 Args:
264 filename (str): The path to the service account json file.
265 kwargs: Additional arguments to pass to the constructor.
267 Returns:
268 google.auth.service_account.Credentials: The constructed
269 credentials.
270 """
271 info, signer = _service_account_info.from_filename(
272 filename, require=["client_email", "token_uri"]
273 )
274 return cls._from_signer_and_info(signer, info, **kwargs)
276 @property
277 def service_account_email(self):
278 """The service account email."""
279 return self._service_account_email
281 @property
282 def project_id(self):
283 """Project ID associated with this credential."""
284 return self._project_id
286 @property
287 def requires_scopes(self):
288 """Checks if the credentials requires scopes.
290 Returns:
291 bool: True if there are no scopes set otherwise False.
292 """
293 return True if not self._scopes else False
295 def _make_copy(self):
296 cred = self.__class__(
297 self._signer,
298 service_account_email=self._service_account_email,
299 scopes=copy.copy(self._scopes),
300 default_scopes=copy.copy(self._default_scopes),
301 token_uri=self._token_uri,
302 subject=self._subject,
303 project_id=self._project_id,
304 quota_project_id=self._quota_project_id,
305 additional_claims=self._additional_claims.copy(),
306 always_use_jwt_access=self._always_use_jwt_access,
307 universe_domain=self._universe_domain,
308 trust_boundary=self._trust_boundary,
309 )
310 cred._cred_file_path = self._cred_file_path
311 self._copy_regional_access_boundary_manager(cred)
312 return cred
314 @_helpers.copy_docstring(credentials.Scoped)
315 def with_scopes(self, scopes, default_scopes=None):
316 cred = self._make_copy()
317 cred._scopes = scopes
318 cred._default_scopes = default_scopes
319 return cred
321 def with_always_use_jwt_access(self, always_use_jwt_access):
322 """Create a copy of these credentials with the specified always_use_jwt_access value.
324 Args:
325 always_use_jwt_access (bool): Whether always use self signed JWT or not.
327 Returns:
328 google.auth.service_account.Credentials: A new credentials
329 instance.
330 Raises:
331 google.auth.exceptions.InvalidValue: If the universe domain is not
332 default and always_use_jwt_access is False.
333 """
334 cred = self._make_copy()
335 if (
336 cred._universe_domain != credentials.DEFAULT_UNIVERSE_DOMAIN
337 and not always_use_jwt_access
338 ):
339 raise exceptions.InvalidValue(
340 "always_use_jwt_access should be True for non-default universe domain"
341 )
342 cred._always_use_jwt_access = always_use_jwt_access
343 return cred
345 @_helpers.copy_docstring(credentials.CredentialsWithUniverseDomain)
346 def with_universe_domain(self, universe_domain):
347 cred = self._make_copy()
348 cred._universe_domain = universe_domain
349 if universe_domain != credentials.DEFAULT_UNIVERSE_DOMAIN:
350 cred._always_use_jwt_access = True
351 return cred
353 def with_subject(self, subject):
354 """Create a copy of these credentials with the specified subject.
356 Args:
357 subject (str): The subject claim.
359 Returns:
360 google.auth.service_account.Credentials: A new credentials
361 instance.
362 """
363 cred = self._make_copy()
364 cred._subject = subject
365 return cred
367 def with_claims(self, additional_claims):
368 """Returns a copy of these credentials with modified claims.
370 Args:
371 additional_claims (Mapping[str, str]): Any additional claims for
372 the JWT payload. This will be merged with the current
373 additional claims.
375 Returns:
376 google.auth.service_account.Credentials: A new credentials
377 instance.
378 """
379 new_additional_claims = copy.deepcopy(self._additional_claims)
380 new_additional_claims.update(additional_claims or {})
381 cred = self._make_copy()
382 cred._additional_claims = new_additional_claims
383 return cred
385 @_helpers.copy_docstring(credentials.CredentialsWithQuotaProject)
386 def with_quota_project(self, quota_project_id):
387 cred = self._make_copy()
388 cred._quota_project_id = quota_project_id
389 return cred
391 @_helpers.copy_docstring(credentials.CredentialsWithTokenUri)
392 def with_token_uri(self, token_uri):
393 cred = self._make_copy()
394 cred._token_uri = token_uri
395 return cred
397 def _make_authorization_grant_assertion(self):
398 """Create the OAuth 2.0 assertion.
400 This assertion is used during the OAuth 2.0 grant to acquire an
401 access token.
403 Returns:
404 bytes: The authorization grant assertion.
405 """
406 now = _helpers.utcnow()
407 lifetime = datetime.timedelta(seconds=_DEFAULT_TOKEN_LIFETIME_SECS)
408 expiry = now + lifetime
410 payload = {
411 "iat": _helpers.datetime_to_secs(now),
412 "exp": _helpers.datetime_to_secs(expiry),
413 # The issuer must be the service account email.
414 "iss": self._service_account_email,
415 # The audience must be the auth token endpoint's URI
416 "aud": _GOOGLE_OAUTH2_TOKEN_ENDPOINT,
417 "scope": _helpers.scopes_to_string(self._scopes or ()),
418 }
420 payload.update(self._additional_claims)
422 # The subject can be a user email for domain-wide delegation.
423 if self._subject:
424 payload.setdefault("sub", self._subject)
426 token = jwt.encode(self._signer, payload)
428 return token
430 def _use_self_signed_jwt(self):
431 # Since domain wide delegation doesn't work with self signed JWT. If
432 # subject exists, then we should not use self signed JWT.
433 return self._subject is None and self._jwt_credentials is not None
435 def _metric_header_for_usage(self):
436 if self._use_self_signed_jwt():
437 return metrics.CRED_TYPE_SA_JWT
438 return metrics.CRED_TYPE_SA_ASSERTION
440 @_helpers.copy_docstring(credentials.CredentialsWithRegionalAccessBoundary)
441 def _perform_refresh_token(self, request):
442 if self._always_use_jwt_access and not self._jwt_credentials:
443 # If self signed jwt should be used but jwt credential is not
444 # created, try to create one with scopes
445 self._create_self_signed_jwt(None)
447 if (
448 self._universe_domain != credentials.DEFAULT_UNIVERSE_DOMAIN
449 and self._subject
450 ):
451 raise exceptions.RefreshError(
452 "domain wide delegation is not supported for non-default universe domain"
453 )
455 if self._use_self_signed_jwt():
456 self._jwt_credentials.refresh(request)
457 self.token = self._jwt_credentials.token.decode()
458 self.expiry = self._jwt_credentials.expiry
459 else:
460 assertion = self._make_authorization_grant_assertion()
461 access_token, expiry, _ = _client.jwt_grant(
462 request, self._token_uri, assertion
463 )
464 self.token = access_token
465 self.expiry = expiry
467 def _create_self_signed_jwt(self, audience):
468 """Create a self-signed JWT from the credentials if requirements are met.
470 Args:
471 audience (str): The service URL. ``https://[API_ENDPOINT]/``
472 """
473 # https://google.aip.dev/auth/4111
474 if self._always_use_jwt_access:
475 if self._scopes:
476 additional_claims = {"scope": " ".join(self._scopes)}
477 if (
478 self._jwt_credentials is None
479 or self._jwt_credentials.additional_claims != additional_claims
480 ):
481 self._jwt_credentials = jwt.Credentials.from_signing_credentials(
482 self, None, additional_claims=additional_claims
483 )
484 elif audience:
485 if (
486 self._jwt_credentials is None
487 or self._jwt_credentials._audience != audience
488 ):
489 self._jwt_credentials = jwt.Credentials.from_signing_credentials(
490 self, audience
491 )
492 elif self._default_scopes:
493 additional_claims = {"scope": " ".join(self._default_scopes)}
494 if (
495 self._jwt_credentials is None
496 or additional_claims != self._jwt_credentials.additional_claims
497 ):
498 self._jwt_credentials = jwt.Credentials.from_signing_credentials(
499 self, None, additional_claims=additional_claims
500 )
501 elif not self._scopes and audience:
502 self._jwt_credentials = jwt.Credentials.from_signing_credentials(
503 self, audience
504 )
506 def _build_regional_access_boundary_lookup_url(
507 self, request: "Optional[google.auth.transport.Request]" = None # noqa: F821
508 ):
509 """Builds and returns the URL for the Regional Access Boundary lookup API.
511 This method constructs the specific URL for the IAM Credentials API's
512 `allowedLocations` endpoint, using the credential's universe domain
513 and service account email.
515 Returns:
516 Optional[str]: The URL for the Regional Access Boundary lookup endpoint, or None
517 if the service account email is missing. Returns None if the subject is populated.
518 """
519 if self._subject:
520 # RAB does not apply to Workspace User Accounts via Domain-wide Delegation.
521 return None
523 if not self.service_account_email:
524 _LOGGER.error(
525 "Service account email is required to build the Regional Access Boundary lookup URL for service account credentials."
526 )
527 return None
528 return _regional_access_boundary_utils.get_service_account_rab_endpoint(
529 self._service_account_email
530 )
532 @_helpers.copy_docstring(credentials.Signing)
533 def sign_bytes(self, message):
534 return self._signer.sign(message)
536 @property # type: ignore
537 @_helpers.copy_docstring(credentials.Signing)
538 def signer(self):
539 return self._signer
541 @property # type: ignore
542 @_helpers.copy_docstring(credentials.Signing)
543 def signer_email(self):
544 return self._service_account_email
546 @_helpers.copy_docstring(credentials.Credentials)
547 def get_cred_info(self):
548 if self._cred_file_path:
549 return {
550 "credential_source": self._cred_file_path,
551 "credential_type": "service account credentials",
552 "principal": self.service_account_email,
553 }
554 return None
557class IDTokenCredentials(
558 credentials.Signing,
559 credentials.CredentialsWithQuotaProject,
560 credentials.CredentialsWithTokenUri,
561):
562 """Open ID Connect ID Token-based service account credentials.
564 These credentials are largely similar to :class:`.Credentials`, but instead
565 of using an OAuth 2.0 Access Token as the bearer token, they use an Open
566 ID Connect ID Token as the bearer token. These credentials are useful when
567 communicating to services that require ID Tokens and can not accept access
568 tokens.
570 Usually, you'll create these credentials with one of the helper
571 constructors. To create credentials using a Google service account
572 private key JSON file::
574 credentials = (
575 service_account.IDTokenCredentials.from_service_account_file(
576 'service-account.json'))
579 Or if you already have the service account file loaded::
581 service_account_info = json.load(open('service_account.json'))
582 credentials = (
583 service_account.IDTokenCredentials.from_service_account_info(
584 service_account_info))
587 Both helper methods pass on arguments to the constructor, so you can
588 specify additional scopes and a subject if necessary::
590 credentials = (
591 service_account.IDTokenCredentials.from_service_account_file(
592 'service-account.json',
593 scopes=['email'],
594 subject='user@example.com'))
597 The credentials are considered immutable. If you want to modify the scopes
598 or the subject used for delegation, use :meth:`with_scopes` or
599 :meth:`with_subject`::
601 scoped_credentials = credentials.with_scopes(['email'])
602 delegated_credentials = credentials.with_subject(subject)
604 """
606 def __init__(
607 self,
608 signer,
609 service_account_email,
610 token_uri,
611 target_audience,
612 additional_claims=None,
613 quota_project_id=None,
614 universe_domain=credentials.DEFAULT_UNIVERSE_DOMAIN,
615 ):
616 """
617 Args:
618 signer (google.auth.crypt.Signer): The signer used to sign JWTs.
619 service_account_email (str): The service account's email.
620 token_uri (str): The OAuth 2.0 Token URI.
621 target_audience (str): The intended audience for these credentials,
622 used when requesting the ID Token. The ID Token's ``aud`` claim
623 will be set to this string.
624 additional_claims (Mapping[str, str]): Any additional claims for
625 the JWT assertion used in the authorization grant.
626 quota_project_id (Optional[str]): The project ID used for quota and billing.
627 universe_domain (str): The universe domain. The default
628 universe domain is googleapis.com. For default value IAM ID
629 token endponint is used for token refresh. Note that
630 iam.serviceAccountTokenCreator role is required to use the IAM
631 endpoint.
633 .. note:: Typically one of the helper constructors
634 :meth:`from_service_account_file` or
635 :meth:`from_service_account_info` are used instead of calling the
636 constructor directly.
637 """
638 super(IDTokenCredentials, self).__init__()
639 self._signer = signer
640 self._service_account_email = service_account_email
641 self._token_uri = token_uri
642 self._target_audience = target_audience
643 self._quota_project_id = quota_project_id
644 self._use_iam_endpoint = False
646 if not universe_domain:
647 self._universe_domain = credentials.DEFAULT_UNIVERSE_DOMAIN
648 else:
649 self._universe_domain = universe_domain
650 self._iam_id_token_endpoint = iam._IAM_IDTOKEN_ENDPOINT.replace(
651 "googleapis.com", self._universe_domain
652 )
654 if self._universe_domain != credentials.DEFAULT_UNIVERSE_DOMAIN:
655 self._use_iam_endpoint = True
657 if additional_claims is not None:
658 self._additional_claims = additional_claims
659 else:
660 self._additional_claims = {}
662 @classmethod
663 def _from_signer_and_info(cls, signer, info, **kwargs):
664 """Creates a credentials instance from a signer and service account
665 info.
667 Args:
668 signer (google.auth.crypt.Signer): The signer used to sign JWTs.
669 info (Mapping[str, str]): The service account info.
670 kwargs: Additional arguments to pass to the constructor.
672 Returns:
673 google.auth.jwt.IDTokenCredentials: The constructed credentials.
675 Raises:
676 ValueError: If the info is not in the expected format.
677 """
678 kwargs.setdefault("service_account_email", info["client_email"])
679 kwargs.setdefault("token_uri", info["token_uri"])
680 if "universe_domain" in info:
681 kwargs["universe_domain"] = info["universe_domain"]
682 return cls(signer, **kwargs)
684 @classmethod
685 def from_service_account_info(cls, info, **kwargs):
686 """Creates a credentials instance from parsed service account info.
688 Args:
689 info (Mapping[str, str]): The service account info in Google
690 format.
691 kwargs: Additional arguments to pass to the constructor.
693 Returns:
694 google.auth.service_account.IDTokenCredentials: The constructed
695 credentials.
697 Raises:
698 ValueError: If the info is not in the expected format.
699 """
700 signer = _service_account_info.from_dict(
701 info, require=["client_email", "token_uri"]
702 )
703 return cls._from_signer_and_info(signer, info, **kwargs)
705 @classmethod
706 def from_service_account_file(cls, filename, **kwargs):
707 """Creates a credentials instance from a service account json file.
709 Args:
710 filename (str): The path to the service account json file.
711 kwargs: Additional arguments to pass to the constructor.
713 Returns:
714 google.auth.service_account.IDTokenCredentials: The constructed
715 credentials.
716 """
717 info, signer = _service_account_info.from_filename(
718 filename, require=["client_email", "token_uri"]
719 )
720 return cls._from_signer_and_info(signer, info, **kwargs)
722 def _make_copy(self):
723 cred = self.__class__(
724 self._signer,
725 service_account_email=self._service_account_email,
726 token_uri=self._token_uri,
727 target_audience=self._target_audience,
728 additional_claims=self._additional_claims.copy(),
729 quota_project_id=self.quota_project_id,
730 universe_domain=self._universe_domain,
731 )
732 # _use_iam_endpoint is not exposed in the constructor
733 cred._use_iam_endpoint = self._use_iam_endpoint
734 return cred
736 def with_target_audience(self, target_audience):
737 """Create a copy of these credentials with the specified target
738 audience.
740 Args:
741 target_audience (str): The intended audience for these credentials,
742 used when requesting the ID Token.
744 Returns:
745 google.auth.service_account.IDTokenCredentials: A new credentials
746 instance.
747 """
748 cred = self._make_copy()
749 cred._target_audience = target_audience
750 return cred
752 def _with_use_iam_endpoint(self, use_iam_endpoint):
753 """Create a copy of these credentials with the use_iam_endpoint value.
755 Args:
756 use_iam_endpoint (bool): If True, IAM generateIdToken endpoint will
757 be used instead of the token_uri. Note that
758 iam.serviceAccountTokenCreator role is required to use the IAM
759 endpoint. The default value is False. This feature is currently
760 experimental and subject to change without notice.
762 Returns:
763 google.auth.service_account.IDTokenCredentials: A new credentials
764 instance.
765 Raises:
766 google.auth.exceptions.InvalidValue: If the universe domain is not
767 default and use_iam_endpoint is False.
768 """
769 cred = self._make_copy()
770 if (
771 cred._universe_domain != credentials.DEFAULT_UNIVERSE_DOMAIN
772 and not use_iam_endpoint
773 ):
774 raise exceptions.InvalidValue(
775 "use_iam_endpoint should be True for non-default universe domain"
776 )
777 cred._use_iam_endpoint = use_iam_endpoint
778 return cred
780 @_helpers.copy_docstring(credentials.CredentialsWithQuotaProject)
781 def with_quota_project(self, quota_project_id):
782 cred = self._make_copy()
783 cred._quota_project_id = quota_project_id
784 return cred
786 @_helpers.copy_docstring(credentials.CredentialsWithTokenUri)
787 def with_token_uri(self, token_uri):
788 cred = self._make_copy()
789 cred._token_uri = token_uri
790 return cred
792 def _make_authorization_grant_assertion(self):
793 """Create the OAuth 2.0 assertion.
795 This assertion is used during the OAuth 2.0 grant to acquire an
796 ID token.
798 Returns:
799 bytes: The authorization grant assertion.
800 """
801 now = _helpers.utcnow()
802 lifetime = datetime.timedelta(seconds=_DEFAULT_TOKEN_LIFETIME_SECS)
803 expiry = now + lifetime
805 payload = {
806 "iat": _helpers.datetime_to_secs(now),
807 "exp": _helpers.datetime_to_secs(expiry),
808 # The issuer must be the service account email.
809 "iss": self.service_account_email,
810 # The audience must be the auth token endpoint's URI
811 "aud": _GOOGLE_OAUTH2_TOKEN_ENDPOINT,
812 # The target audience specifies which service the ID token is
813 # intended for.
814 "target_audience": self._target_audience,
815 }
817 payload.update(self._additional_claims)
819 token = jwt.encode(self._signer, payload)
821 return token
823 def _refresh_with_iam_endpoint(self, request):
824 """Use IAM generateIdToken endpoint to obtain an ID token.
826 It works as follows:
828 1. First we create a self signed jwt with
829 https://www.googleapis.com/auth/iam being the scope.
831 2. Next we use the self signed jwt as the access token, and make a POST
832 request to IAM generateIdToken endpoint. The request body is:
833 {
834 "audience": self._target_audience,
835 "includeEmail": "true",
836 "useEmailAzp": "true",
837 }
839 If the request is succesfully, it will return {"token":"the ID token"},
840 and we can extract the ID token and compute its expiry.
841 """
842 jwt_credentials = jwt.Credentials.from_signing_credentials(
843 self,
844 None,
845 additional_claims={"scope": "https://www.googleapis.com/auth/iam"},
846 )
847 jwt_credentials.refresh(request)
849 self.token, self.expiry = _client.call_iam_generate_id_token_endpoint(
850 request,
851 self._iam_id_token_endpoint,
852 self.signer_email,
853 self._target_audience,
854 jwt_credentials.token.decode(),
855 self._universe_domain,
856 )
858 @_helpers.copy_docstring(credentials.Credentials)
859 def refresh(self, request):
860 if self._use_iam_endpoint:
861 self._refresh_with_iam_endpoint(request)
862 else:
863 assertion = self._make_authorization_grant_assertion()
864 access_token, expiry, _ = _client.id_token_jwt_grant(
865 request, self._token_uri, assertion
866 )
867 self.token = access_token
868 self.expiry = expiry
870 @property
871 def service_account_email(self):
872 """The service account email."""
873 return self._service_account_email
875 @_helpers.copy_docstring(credentials.Signing)
876 def sign_bytes(self, message):
877 return self._signer.sign(message)
879 @property # type: ignore
880 @_helpers.copy_docstring(credentials.Signing)
881 def signer(self):
882 return self._signer
884 @property # type: ignore
885 @_helpers.copy_docstring(credentials.Signing)
886 def signer_email(self):
887 return self._service_account_email