Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.11/site-packages/google/auth/jwt.py: 29%
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"""JSON Web Tokens
17Provides support for creating (encoding) and verifying (decoding) JWTs,
18especially JWTs generated and consumed by Google infrastructure.
20See `rfc7519`_ for more details on JWTs.
22To encode a JWT use :func:`encode`::
24 from google.auth import crypt
25 from google.auth import jwt
27 signer = crypt.Signer(private_key)
28 payload = {'some': 'payload'}
29 encoded = jwt.encode(signer, payload)
31To decode a JWT and verify claims use :func:`decode`::
33 claims = jwt.decode(encoded, certs=public_certs)
35You can also skip verification::
37 claims = jwt.decode(encoded, verify=False)
39.. _rfc7519: https://tools.ietf.org/html/rfc7519
41"""
43try:
44 from collections.abc import Iterable, Mapping
45# Python 2.7 compatibility
46except ImportError: # pragma: NO COVER
47 from collections import Iterable, Mapping # type: ignore
48import copy
49import datetime
50import json
51import urllib
53import google.auth.credentials
54from google.auth import (
55 _cache,
56 _helpers,
57 _regional_access_boundary_utils,
58 _service_account_info,
59 crypt,
60 exceptions,
61)
63try:
64 from google.auth.crypt import es
65except ImportError: # pragma: NO COVER
66 es = None # type: ignore
68_DEFAULT_TOKEN_LIFETIME_SECS = 3600 # 1 hour in seconds
69_DEFAULT_MAX_CACHE_SIZE = 10
70_ALGORITHM_TO_VERIFIER_CLASS = {"RS256": crypt.RSAVerifier}
71_CRYPTOGRAPHY_BASED_ALGORITHMS = frozenset(["ES256", "ES384"])
73if es is not None: # pragma: NO COVER
74 _ALGORITHM_TO_VERIFIER_CLASS["ES256"] = es.EsVerifier # type: ignore
75 _ALGORITHM_TO_VERIFIER_CLASS["ES384"] = es.EsVerifier # type: ignore
78def encode(signer, payload, header=None, key_id=None):
79 """Make a signed JWT.
81 Args:
82 signer (google.auth.crypt.Signer): The signer used to sign the JWT.
83 payload (Mapping[str, str]): The JWT payload.
84 header (Mapping[str, str]): Additional JWT header payload.
85 key_id (str): The key id to add to the JWT header. If the
86 signer has a key id it will be used as the default. If this is
87 specified it will override the signer's key id.
89 Returns:
90 bytes: The encoded JWT.
91 """
92 if header is None:
93 header = {}
95 if key_id is None:
96 key_id = signer.key_id
98 header.update({"typ": "JWT"})
100 if "alg" not in header:
101 if es is not None and isinstance(signer, es.EsSigner):
102 header.update({"alg": signer.algorithm})
103 else:
104 header.update({"alg": "RS256"})
106 if key_id is not None:
107 header["kid"] = key_id
109 segments = [
110 _helpers.unpadded_urlsafe_b64encode(json.dumps(header).encode("utf-8")),
111 _helpers.unpadded_urlsafe_b64encode(json.dumps(payload).encode("utf-8")),
112 ]
114 signing_input = b".".join(segments)
115 signature = signer.sign(signing_input)
116 segments.append(_helpers.unpadded_urlsafe_b64encode(signature))
118 return b".".join(segments)
121def _decode_jwt_segment(encoded_section):
122 """Decodes a single JWT segment."""
123 section_bytes = _helpers.padded_urlsafe_b64decode(encoded_section)
124 try:
125 return json.loads(section_bytes.decode("utf-8"))
126 except ValueError as caught_exc:
127 new_exc = exceptions.MalformedError(
128 "Can't parse segment: {0}".format(section_bytes)
129 )
130 raise new_exc from caught_exc
133def _unverified_decode(token):
134 """Decodes a token and does no verification.
136 Args:
137 token (Union[str, bytes]): The encoded JWT.
139 Returns:
140 Tuple[Mapping, Mapping, str, str]: header, payload, signed_section, and
141 signature.
143 Raises:
144 google.auth.exceptions.MalformedError: if there are an incorrect amount of segments in the token or segments of the wrong type.
145 """
146 token = _helpers.to_bytes(token)
148 if token.count(b".") != 2:
149 raise exceptions.MalformedError(
150 "Wrong number of segments in token: {0}".format(token)
151 )
153 encoded_header, encoded_payload, signature = token.split(b".")
154 signed_section = encoded_header + b"." + encoded_payload
155 signature = _helpers.padded_urlsafe_b64decode(signature)
157 # Parse segments
158 header = _decode_jwt_segment(encoded_header)
159 payload = _decode_jwt_segment(encoded_payload)
161 if not isinstance(header, Mapping):
162 raise exceptions.MalformedError(
163 "Header segment should be a JSON object: {0}".format(encoded_header)
164 )
166 if not isinstance(payload, Mapping):
167 raise exceptions.MalformedError(
168 "Payload segment should be a JSON object: {0}".format(encoded_payload)
169 )
171 return header, payload, signed_section, signature
174def decode_header(token):
175 """Return the decoded header of a token.
177 No verification is done. This is useful to extract the key id from
178 the header in order to acquire the appropriate certificate to verify
179 the token.
181 Args:
182 token (Union[str, bytes]): the encoded JWT.
184 Returns:
185 Mapping: The decoded JWT header.
186 """
187 header, _, _, _ = _unverified_decode(token)
188 return header
191def _verify_iat_and_exp(payload, clock_skew_in_seconds=0):
192 """Verifies the ``iat`` (Issued At) and ``exp`` (Expires) claims in a token
193 payload.
195 Args:
196 payload (Mapping[str, str]): The JWT payload.
197 clock_skew_in_seconds (int): The clock skew used for `iat` and `exp`
198 validation.
200 Raises:
201 google.auth.exceptions.InvalidValue: if value validation failed.
202 google.auth.exceptions.MalformedError: if schema validation failed.
203 """
204 now = _helpers.datetime_to_secs(_helpers.utcnow())
206 # Make sure the iat and exp claims are present.
207 for key in ("iat", "exp"):
208 if key not in payload:
209 raise exceptions.MalformedError(
210 "Token does not contain required claim {}".format(key)
211 )
213 # Make sure the token wasn't issued in the future.
214 iat = payload["iat"]
215 # Err on the side of accepting a token that is slightly early to account
216 # for clock skew.
217 earliest = iat - clock_skew_in_seconds
218 if now < earliest:
219 raise exceptions.InvalidValue(
220 "Token used too early, {} < {}. Check that your computer's clock is set correctly.".format(
221 now, iat
222 )
223 )
225 # Make sure the token wasn't issued in the past.
226 exp = payload["exp"]
227 # Err on the side of accepting a token that is slightly out of date
228 # to account for clow skew.
229 latest = exp + clock_skew_in_seconds
230 if latest < now:
231 raise exceptions.InvalidValue("Token expired, {} < {}".format(latest, now))
234def decode(token, certs=None, verify=True, audience=None, clock_skew_in_seconds=0):
235 """Decode and verify a JWT.
237 Args:
238 token (str): The encoded JWT.
239 certs (Union[str, bytes, Mapping[str, Union[str, bytes]]]): The
240 certificate used to validate the JWT signature. If bytes or string,
241 it must the the public key certificate in PEM format. If a mapping,
242 it must be a mapping of key IDs to public key certificates in PEM
243 format. The mapping must contain the same key ID that's specified
244 in the token's header.
245 verify (bool): Whether to perform signature and claim validation.
246 Verification is done by default.
247 audience (str or list): The audience claim, 'aud', that this JWT should
248 contain. Or a list of audience claims. If None then the JWT's 'aud'
249 parameter is not verified.
250 clock_skew_in_seconds (int): The clock skew used for `iat` and `exp`
251 validation.
253 Returns:
254 Mapping[str, str]: The deserialized JSON payload in the JWT.
256 Raises:
257 google.auth.exceptions.InvalidValue: if value validation failed.
258 google.auth.exceptions.MalformedError: if schema validation failed.
259 """
260 header, payload, signed_section, signature = _unverified_decode(token)
262 if not verify:
263 return payload
265 # Pluck the key id and algorithm from the header and make sure we have
266 # a verifier that can support it.
267 key_alg = header.get("alg")
268 key_id = header.get("kid")
270 try:
271 verifier_cls = _ALGORITHM_TO_VERIFIER_CLASS[key_alg]
272 except (KeyError, TypeError) as exc:
273 if isinstance(key_alg, str) and key_alg in _CRYPTOGRAPHY_BASED_ALGORITHMS:
274 raise exceptions.InvalidValue(
275 "The key algorithm {} requires the cryptography package to be installed.".format(
276 key_alg
277 )
278 ) from exc
279 else:
280 raise exceptions.InvalidValue(
281 "Unsupported signature algorithm {}".format(key_alg)
282 ) from exc
283 # If certs is specified as a dictionary of key IDs to certificates, then
284 # use the certificate identified by the key ID in the token header.
285 if isinstance(certs, Mapping):
286 if key_id:
287 try:
288 certs_to_check = [certs[key_id]]
289 except (KeyError, TypeError) as exc:
290 raise exceptions.MalformedError(
291 "Certificate for key id {} not found.".format(key_id)
292 ) from exc
293 # If there's no key id in the header, check against all of the certs.
294 else:
295 certs_to_check = certs.values()
296 else:
297 certs_to_check = certs
299 # Verify that the signature matches the message.
300 try:
301 valid_signature = crypt.verify_signature(
302 signed_section, signature, certs_to_check, verifier_cls
303 )
304 except TypeError as exc:
305 if not isinstance(certs_to_check, Iterable):
306 raise
307 raise exceptions.InvalidValue(
308 "Certificate key type does not match signature algorithm {}.".format(
309 key_alg
310 )
311 ) from exc
312 if not valid_signature:
313 raise exceptions.MalformedError("Could not verify token signature.")
315 # Verify the issued at and created times in the payload.
316 _verify_iat_and_exp(payload, clock_skew_in_seconds)
318 # Check audience.
319 if audience is not None:
320 claim_audience = payload.get("aud")
321 if isinstance(audience, str):
322 audience = [audience]
323 if claim_audience not in audience:
324 raise exceptions.InvalidValue(
325 "Token has wrong audience {}, expected one of {}".format(
326 claim_audience, audience
327 )
328 )
330 return payload
333class Credentials(
334 google.auth.credentials.Signing,
335 google.auth.credentials.CredentialsWithQuotaProject,
336 google.auth.credentials.CredentialsWithRegionalAccessBoundary,
337):
338 """Credentials that use a JWT as the bearer token.
340 These credentials require an "audience" claim. This claim identifies the
341 intended recipient of the bearer token.
343 The constructor arguments determine the claims for the JWT that is
344 sent with requests. Usually, you'll construct these credentials with
345 one of the helper constructors as shown in the next section.
347 To create JWT credentials using a Google service account private key
348 JSON file::
350 audience = 'https://pubsub.googleapis.com/google.pubsub.v1.Publisher'
351 credentials = jwt.Credentials.from_service_account_file(
352 'service-account.json',
353 audience=audience)
355 If you already have the service account file loaded and parsed::
357 service_account_info = json.load(open('service_account.json'))
358 credentials = jwt.Credentials.from_service_account_info(
359 service_account_info,
360 audience=audience)
362 Both helper methods pass on arguments to the constructor, so you can
363 specify the JWT claims::
365 credentials = jwt.Credentials.from_service_account_file(
366 'service-account.json',
367 audience=audience,
368 additional_claims={'meta': 'data'})
370 You can also construct the credentials directly if you have a
371 :class:`~google.auth.crypt.Signer` instance::
373 credentials = jwt.Credentials(
374 signer,
375 issuer='your-issuer',
376 subject='your-subject',
377 audience=audience)
379 The claims are considered immutable. If you want to modify the claims,
380 you can easily create another instance using :meth:`with_claims`::
382 new_audience = (
383 'https://pubsub.googleapis.com/google.pubsub.v1.Subscriber')
384 new_credentials = credentials.with_claims(audience=new_audience)
385 """
387 def __init__(
388 self,
389 signer,
390 issuer,
391 subject,
392 audience,
393 additional_claims=None,
394 token_lifetime=_DEFAULT_TOKEN_LIFETIME_SECS,
395 quota_project_id=None,
396 ):
397 """
398 Args:
399 signer (google.auth.crypt.Signer): The signer used to sign JWTs.
400 issuer (str): The `iss` claim.
401 subject (str): The `sub` claim.
402 audience (str): the `aud` claim. The intended audience for the
403 credentials.
404 additional_claims (Mapping[str, str]): Any additional claims for
405 the JWT payload.
406 token_lifetime (int): The amount of time in seconds for
407 which the token is valid. Defaults to 1 hour.
408 quota_project_id (Optional[str]): The project ID used for quota
409 and billing.
410 """
411 super(Credentials, self).__init__()
412 self._signer = signer
413 self._issuer = issuer
414 self._subject = subject
415 self._audience = audience
416 self._token_lifetime = token_lifetime
417 self._quota_project_id = quota_project_id
419 if additional_claims is None:
420 additional_claims = {}
422 self._additional_claims = additional_claims
424 @classmethod
425 def _from_signer_and_info(cls, signer, info, **kwargs):
426 """Creates a Credentials instance from a signer and service account
427 info.
429 Args:
430 signer (google.auth.crypt.Signer): The signer used to sign JWTs.
431 info (Mapping[str, str]): The service account info.
432 kwargs: Additional arguments to pass to the constructor.
434 Returns:
435 google.auth.jwt.Credentials: The constructed credentials.
437 Raises:
438 google.auth.exceptions.MalformedError: If the info is not in the expected format.
439 """
440 kwargs.setdefault("subject", info["client_email"])
441 kwargs.setdefault("issuer", info["client_email"])
442 return cls(signer, **kwargs)
444 @classmethod
445 def from_service_account_info(cls, info, **kwargs):
446 """Creates an Credentials instance from a dictionary.
448 Args:
449 info (Mapping[str, str]): The service account info in Google
450 format.
451 kwargs: Additional arguments to pass to the constructor.
453 Returns:
454 google.auth.jwt.Credentials: The constructed credentials.
456 Raises:
457 google.auth.exceptions.MalformedError: If the info is not in the expected format.
458 """
459 signer = _service_account_info.from_dict(info, require=["client_email"])
460 return cls._from_signer_and_info(signer, info, **kwargs)
462 @classmethod
463 def from_service_account_file(cls, filename, **kwargs):
464 """Creates a Credentials instance from a service account .json file
465 in Google format.
467 Args:
468 filename (str): The path to the service account .json file.
469 kwargs: Additional arguments to pass to the constructor.
471 Returns:
472 google.auth.jwt.Credentials: The constructed credentials.
473 """
474 info, signer = _service_account_info.from_filename(
475 filename, require=["client_email"]
476 )
477 return cls._from_signer_and_info(signer, info, **kwargs)
479 @classmethod
480 def from_signing_credentials(cls, credentials, audience, **kwargs):
481 """Creates a new :class:`google.auth.jwt.Credentials` instance from an
482 existing :class:`google.auth.credentials.Signing` instance.
484 The new instance will use the same signer as the existing instance and
485 will use the existing instance's signer email as the issuer and
486 subject by default.
488 Example::
490 svc_creds = service_account.Credentials.from_service_account_file(
491 'service_account.json')
492 audience = (
493 'https://pubsub.googleapis.com/google.pubsub.v1.Publisher')
494 jwt_creds = jwt.Credentials.from_signing_credentials(
495 svc_creds, audience=audience)
497 Args:
498 credentials (google.auth.credentials.Signing): The credentials to
499 use to construct the new credentials.
500 audience (str): the `aud` claim. The intended audience for the
501 credentials.
502 kwargs: Additional arguments to pass to the constructor.
504 Returns:
505 google.auth.jwt.Credentials: A new Credentials instance.
506 """
507 kwargs.setdefault("issuer", credentials.signer_email)
508 kwargs.setdefault("subject", credentials.signer_email)
509 jwt_creds = cls(credentials.signer, audience=audience, **kwargs)
511 if isinstance(
512 credentials,
513 google.auth.credentials.CredentialsWithRegionalAccessBoundary,
514 ):
515 credentials._copy_regional_access_boundary_manager(jwt_creds)
517 return jwt_creds
519 def with_claims(
520 self, issuer=None, subject=None, audience=None, additional_claims=None
521 ):
522 """Returns a copy of these credentials with modified claims.
524 Args:
525 issuer (str): The `iss` claim. If unspecified the current issuer
526 claim will be used.
527 subject (str): The `sub` claim. If unspecified the current subject
528 claim will be used.
529 audience (str): the `aud` claim. If unspecified the current
530 audience claim will be used.
531 additional_claims (Mapping[str, str]): Any additional claims for
532 the JWT payload. This will be merged with the current
533 additional claims.
535 Returns:
536 google.auth.jwt.Credentials: A new credentials instance.
537 """
538 new_additional_claims = copy.deepcopy(self._additional_claims)
539 new_additional_claims.update(additional_claims or {})
541 cred = self.__class__(
542 self._signer,
543 issuer=issuer if issuer is not None else self._issuer,
544 subject=subject if subject is not None else self._subject,
545 audience=audience if audience is not None else self._audience,
546 additional_claims=new_additional_claims,
547 quota_project_id=self._quota_project_id,
548 )
549 self._copy_regional_access_boundary_manager(cred)
550 return cred
552 @_helpers.copy_docstring(google.auth.credentials.CredentialsWithQuotaProject)
553 def with_quota_project(self, quota_project_id):
554 cred = self.__class__(
555 self._signer,
556 issuer=self._issuer,
557 subject=self._subject,
558 audience=self._audience,
559 additional_claims=self._additional_claims,
560 quota_project_id=quota_project_id,
561 )
562 self._copy_regional_access_boundary_manager(cred)
563 return cred
565 def _make_jwt(self):
566 """Make a signed JWT.
568 Returns:
569 Tuple[bytes, datetime]: The encoded JWT and the expiration.
570 """
571 now = _helpers.utcnow()
572 lifetime = datetime.timedelta(seconds=self._token_lifetime)
573 expiry = now + lifetime
575 payload = {
576 "iss": self._issuer,
577 "sub": self._subject,
578 "iat": _helpers.datetime_to_secs(now),
579 "exp": _helpers.datetime_to_secs(expiry),
580 }
581 if self._audience:
582 payload["aud"] = self._audience
584 payload.update(self._additional_claims)
586 jwt = encode(self._signer, payload)
588 return jwt, expiry
590 def _perform_refresh_token(self, request):
591 """Refreshes the access token.
593 Args:
594 request (Any): Unused.
595 """
596 # pylint: disable=unused-argument
597 # (pylint doesn't correctly recognize overridden methods.)
598 self.token, self.expiry = self._make_jwt()
600 def _build_regional_access_boundary_lookup_url(self, request=None):
601 """Builds the lookup URL using the service account's email address.
603 Returns None if the subject is populated.
604 """
605 # In jwt.Credentials, subject defaults to client_email (which is the issuer).
606 # We must check self._subject != self._issuer to correctly determine if
607 # Domain-Wide Delegation is active.
608 if self._subject and self._subject != self._issuer:
609 # RAB does not apply to Workspace User Accounts via Domain-wide Delegation.
610 return None
612 if not self.signer_email:
613 return None
615 return _regional_access_boundary_utils.get_service_account_rab_endpoint(
616 self.signer_email
617 )
619 @_helpers.copy_docstring(google.auth.credentials.Signing)
620 def sign_bytes(self, message):
621 return self._signer.sign(message)
623 @property # type: ignore
624 @_helpers.copy_docstring(google.auth.credentials.Signing)
625 def signer_email(self):
626 return self._issuer
628 @property # type: ignore
629 @_helpers.copy_docstring(google.auth.credentials.Signing)
630 def signer(self):
631 return self._signer
633 @property # type: ignore
634 def additional_claims(self):
635 """Additional claims the JWT object was created with."""
636 return self._additional_claims
639class OnDemandCredentials(
640 google.auth.credentials.Signing, google.auth.credentials.CredentialsWithQuotaProject
641):
642 """On-demand JWT credentials.
644 Like :class:`Credentials`, this class uses a JWT as the bearer token for
645 authentication. However, this class does not require the audience at
646 construction time. Instead, it will generate a new token on-demand for
647 each request using the request URI as the audience. It caches tokens
648 so that multiple requests to the same URI do not incur the overhead
649 of generating a new token every time.
651 This behavior is especially useful for `gRPC`_ clients. A gRPC service may
652 have multiple audience and gRPC clients may not know all of the audiences
653 required for accessing a particular service. With these credentials,
654 no knowledge of the audiences is required ahead of time.
656 .. _grpc: http://www.grpc.io/
657 """
659 def __init__(
660 self,
661 signer,
662 issuer,
663 subject,
664 additional_claims=None,
665 token_lifetime=_DEFAULT_TOKEN_LIFETIME_SECS,
666 max_cache_size=_DEFAULT_MAX_CACHE_SIZE,
667 quota_project_id=None,
668 ):
669 """
670 Args:
671 signer (google.auth.crypt.Signer): The signer used to sign JWTs.
672 issuer (str): The `iss` claim.
673 subject (str): The `sub` claim.
674 additional_claims (Mapping[str, str]): Any additional claims for
675 the JWT payload.
676 token_lifetime (int): The amount of time in seconds for
677 which the token is valid. Defaults to 1 hour.
678 max_cache_size (int): The maximum number of JWT tokens to keep in
679 cache. Tokens are cached using :class:`google.auth._cache.LRUCache`.
680 quota_project_id (Optional[str]): The project ID used for quota
681 and billing.
683 """
684 super(OnDemandCredentials, self).__init__()
685 self._signer = signer
686 self._issuer = issuer
687 self._subject = subject
688 self._token_lifetime = token_lifetime
689 self._quota_project_id = quota_project_id
691 if additional_claims is None:
692 additional_claims = {}
694 self._additional_claims = additional_claims
695 self._cache = _cache.LRUCache(maxsize=max_cache_size)
697 @classmethod
698 def _from_signer_and_info(cls, signer, info, **kwargs):
699 """Creates an OnDemandCredentials instance from a signer and service
700 account info.
702 Args:
703 signer (google.auth.crypt.Signer): The signer used to sign JWTs.
704 info (Mapping[str, str]): The service account info.
705 kwargs: Additional arguments to pass to the constructor.
707 Returns:
708 google.auth.jwt.OnDemandCredentials: The constructed credentials.
710 Raises:
711 google.auth.exceptions.MalformedError: If the info is not in the expected format.
712 """
713 kwargs.setdefault("subject", info["client_email"])
714 kwargs.setdefault("issuer", info["client_email"])
715 return cls(signer, **kwargs)
717 @classmethod
718 def from_service_account_info(cls, info, **kwargs):
719 """Creates an OnDemandCredentials instance from a dictionary.
721 Args:
722 info (Mapping[str, str]): The service account info in Google
723 format.
724 kwargs: Additional arguments to pass to the constructor.
726 Returns:
727 google.auth.jwt.OnDemandCredentials: The constructed credentials.
729 Raises:
730 google.auth.exceptions.MalformedError: If the info is not in the expected format.
731 """
732 signer = _service_account_info.from_dict(info, require=["client_email"])
733 return cls._from_signer_and_info(signer, info, **kwargs)
735 @classmethod
736 def from_service_account_file(cls, filename, **kwargs):
737 """Creates an OnDemandCredentials instance from a service account .json
738 file in Google format.
740 Args:
741 filename (str): The path to the service account .json file.
742 kwargs: Additional arguments to pass to the constructor.
744 Returns:
745 google.auth.jwt.OnDemandCredentials: The constructed credentials.
746 """
747 info, signer = _service_account_info.from_filename(
748 filename, require=["client_email"]
749 )
750 return cls._from_signer_and_info(signer, info, **kwargs)
752 @classmethod
753 def from_signing_credentials(cls, credentials, **kwargs):
754 """Creates a new :class:`google.auth.jwt.OnDemandCredentials` instance
755 from an existing :class:`google.auth.credentials.Signing` instance.
757 The new instance will use the same signer as the existing instance and
758 will use the existing instance's signer email as the issuer and
759 subject by default.
761 Example::
763 svc_creds = service_account.Credentials.from_service_account_file(
764 'service_account.json')
765 jwt_creds = jwt.OnDemandCredentials.from_signing_credentials(
766 svc_creds)
768 Args:
769 credentials (google.auth.credentials.Signing): The credentials to
770 use to construct the new credentials.
771 kwargs: Additional arguments to pass to the constructor.
773 Returns:
774 google.auth.jwt.Credentials: A new Credentials instance.
775 """
776 kwargs.setdefault("issuer", credentials.signer_email)
777 kwargs.setdefault("subject", credentials.signer_email)
778 return cls(credentials.signer, **kwargs)
780 def with_claims(self, issuer=None, subject=None, additional_claims=None):
781 """Returns a copy of these credentials with modified claims.
783 Args:
784 issuer (str): The `iss` claim. If unspecified the current issuer
785 claim will be used.
786 subject (str): The `sub` claim. If unspecified the current subject
787 claim will be used.
788 additional_claims (Mapping[str, str]): Any additional claims for
789 the JWT payload. This will be merged with the current
790 additional claims.
792 Returns:
793 google.auth.jwt.OnDemandCredentials: A new credentials instance.
794 """
795 new_additional_claims = copy.deepcopy(self._additional_claims)
796 new_additional_claims.update(additional_claims or {})
798 return self.__class__(
799 self._signer,
800 issuer=issuer if issuer is not None else self._issuer,
801 subject=subject if subject is not None else self._subject,
802 additional_claims=new_additional_claims,
803 max_cache_size=self._cache.maxsize,
804 quota_project_id=self._quota_project_id,
805 )
807 @_helpers.copy_docstring(google.auth.credentials.CredentialsWithQuotaProject)
808 def with_quota_project(self, quota_project_id):
809 return self.__class__(
810 self._signer,
811 issuer=self._issuer,
812 subject=self._subject,
813 additional_claims=self._additional_claims,
814 max_cache_size=self._cache.maxsize,
815 quota_project_id=quota_project_id,
816 )
818 @property
819 def valid(self):
820 """Checks the validity of the credentials.
822 These credentials are always valid because it generates tokens on
823 demand.
824 """
825 return True
827 def _make_jwt_for_audience(self, audience):
828 """Make a new JWT for the given audience.
830 Args:
831 audience (str): The intended audience.
833 Returns:
834 Tuple[bytes, datetime]: The encoded JWT and the expiration.
835 """
836 now = _helpers.utcnow()
837 lifetime = datetime.timedelta(seconds=self._token_lifetime)
838 expiry = now + lifetime
840 payload = {
841 "iss": self._issuer,
842 "sub": self._subject,
843 "iat": _helpers.datetime_to_secs(now),
844 "exp": _helpers.datetime_to_secs(expiry),
845 "aud": audience,
846 }
848 payload.update(self._additional_claims)
850 jwt = encode(self._signer, payload)
852 return jwt, expiry
854 def _get_jwt_for_audience(self, audience):
855 """Get a JWT For a given audience.
857 If there is already an existing, non-expired token in the cache for
858 the audience, that token is used. Otherwise, a new token will be
859 created.
861 Args:
862 audience (str): The intended audience.
864 Returns:
865 bytes: The encoded JWT.
866 """
867 token, expiry = self._cache.get(audience, (None, None))
869 if token is None or expiry < _helpers.utcnow():
870 token, expiry = self._make_jwt_for_audience(audience)
871 self._cache[audience] = token, expiry
873 return token
875 def refresh(self, request):
876 """Raises an exception, these credentials can not be directly
877 refreshed.
879 Args:
880 request (Any): Unused.
882 Raises:
883 google.auth.RefreshError
884 """
885 # pylint: disable=unused-argument
886 # (pylint doesn't correctly recognize overridden methods.)
887 raise exceptions.RefreshError(
888 "OnDemandCredentials can not be directly refreshed."
889 )
891 def before_request(self, request, method, url, headers):
892 """Performs credential-specific before request logic.
894 Args:
895 request (Any): Unused. JWT credentials do not need to make an
896 HTTP request to refresh.
897 method (str): The request's HTTP method.
898 url (str): The request's URI. This is used as the audience claim
899 when generating the JWT.
900 headers (Mapping): The request's headers.
901 """
902 # pylint: disable=unused-argument
903 # (pylint doesn't correctly recognize overridden methods.)
904 parts = urllib.parse.urlsplit(url)
905 # Strip query string and fragment
906 audience = urllib.parse.urlunsplit(
907 (parts.scheme, parts.netloc, parts.path, "", "")
908 )
909 token = self._get_jwt_for_audience(audience)
910 self.apply(headers, token=token)
912 @_helpers.copy_docstring(google.auth.credentials.Signing)
913 def sign_bytes(self, message):
914 return self._signer.sign(message)
916 @property # type: ignore
917 @_helpers.copy_docstring(google.auth.credentials.Signing)
918 def signer_email(self):
919 return self._issuer
921 @property # type: ignore
922 @_helpers.copy_docstring(google.auth.credentials.Signing)
923 def signer(self):
924 return self._signer