Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.11/site-packages/google/auth/jwt.py: 30%
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 Mapping
45# Python 2.7 compatibility
46except ImportError: # pragma: NO COVER
47 from collections import 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 as exc:
273 if 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 if key_id not in certs:
288 raise exceptions.MalformedError(
289 "Certificate for key id {} not found.".format(key_id)
290 )
291 certs_to_check = [certs[key_id]]
292 # If there's no key id in the header, check against all of the certs.
293 else:
294 certs_to_check = certs.values()
295 else:
296 certs_to_check = certs
298 # Verify that the signature matches the message.
299 if not crypt.verify_signature(
300 signed_section, signature, certs_to_check, verifier_cls
301 ):
302 raise exceptions.MalformedError("Could not verify token signature.")
304 # Verify the issued at and created times in the payload.
305 _verify_iat_and_exp(payload, clock_skew_in_seconds)
307 # Check audience.
308 if audience is not None:
309 claim_audience = payload.get("aud")
310 if isinstance(audience, str):
311 audience = [audience]
312 if claim_audience not in audience:
313 raise exceptions.InvalidValue(
314 "Token has wrong audience {}, expected one of {}".format(
315 claim_audience, audience
316 )
317 )
319 return payload
322class Credentials(
323 google.auth.credentials.Signing,
324 google.auth.credentials.CredentialsWithQuotaProject,
325 google.auth.credentials.CredentialsWithRegionalAccessBoundary,
326):
327 """Credentials that use a JWT as the bearer token.
329 These credentials require an "audience" claim. This claim identifies the
330 intended recipient of the bearer token.
332 The constructor arguments determine the claims for the JWT that is
333 sent with requests. Usually, you'll construct these credentials with
334 one of the helper constructors as shown in the next section.
336 To create JWT credentials using a Google service account private key
337 JSON file::
339 audience = 'https://pubsub.googleapis.com/google.pubsub.v1.Publisher'
340 credentials = jwt.Credentials.from_service_account_file(
341 'service-account.json',
342 audience=audience)
344 If you already have the service account file loaded and parsed::
346 service_account_info = json.load(open('service_account.json'))
347 credentials = jwt.Credentials.from_service_account_info(
348 service_account_info,
349 audience=audience)
351 Both helper methods pass on arguments to the constructor, so you can
352 specify the JWT claims::
354 credentials = jwt.Credentials.from_service_account_file(
355 'service-account.json',
356 audience=audience,
357 additional_claims={'meta': 'data'})
359 You can also construct the credentials directly if you have a
360 :class:`~google.auth.crypt.Signer` instance::
362 credentials = jwt.Credentials(
363 signer,
364 issuer='your-issuer',
365 subject='your-subject',
366 audience=audience)
368 The claims are considered immutable. If you want to modify the claims,
369 you can easily create another instance using :meth:`with_claims`::
371 new_audience = (
372 'https://pubsub.googleapis.com/google.pubsub.v1.Subscriber')
373 new_credentials = credentials.with_claims(audience=new_audience)
374 """
376 def __init__(
377 self,
378 signer,
379 issuer,
380 subject,
381 audience,
382 additional_claims=None,
383 token_lifetime=_DEFAULT_TOKEN_LIFETIME_SECS,
384 quota_project_id=None,
385 ):
386 """
387 Args:
388 signer (google.auth.crypt.Signer): The signer used to sign JWTs.
389 issuer (str): The `iss` claim.
390 subject (str): The `sub` claim.
391 audience (str): the `aud` claim. The intended audience for the
392 credentials.
393 additional_claims (Mapping[str, str]): Any additional claims for
394 the JWT payload.
395 token_lifetime (int): The amount of time in seconds for
396 which the token is valid. Defaults to 1 hour.
397 quota_project_id (Optional[str]): The project ID used for quota
398 and billing.
399 """
400 super(Credentials, self).__init__()
401 self._signer = signer
402 self._issuer = issuer
403 self._subject = subject
404 self._audience = audience
405 self._token_lifetime = token_lifetime
406 self._quota_project_id = quota_project_id
408 if additional_claims is None:
409 additional_claims = {}
411 self._additional_claims = additional_claims
413 @classmethod
414 def _from_signer_and_info(cls, signer, info, **kwargs):
415 """Creates a Credentials instance from a signer and service account
416 info.
418 Args:
419 signer (google.auth.crypt.Signer): The signer used to sign JWTs.
420 info (Mapping[str, str]): The service account info.
421 kwargs: Additional arguments to pass to the constructor.
423 Returns:
424 google.auth.jwt.Credentials: The constructed credentials.
426 Raises:
427 google.auth.exceptions.MalformedError: If the info is not in the expected format.
428 """
429 kwargs.setdefault("subject", info["client_email"])
430 kwargs.setdefault("issuer", info["client_email"])
431 return cls(signer, **kwargs)
433 @classmethod
434 def from_service_account_info(cls, info, **kwargs):
435 """Creates an Credentials instance from a dictionary.
437 Args:
438 info (Mapping[str, str]): The service account info in Google
439 format.
440 kwargs: Additional arguments to pass to the constructor.
442 Returns:
443 google.auth.jwt.Credentials: The constructed credentials.
445 Raises:
446 google.auth.exceptions.MalformedError: If the info is not in the expected format.
447 """
448 signer = _service_account_info.from_dict(info, require=["client_email"])
449 return cls._from_signer_and_info(signer, info, **kwargs)
451 @classmethod
452 def from_service_account_file(cls, filename, **kwargs):
453 """Creates a Credentials instance from a service account .json file
454 in Google format.
456 Args:
457 filename (str): The path to the service account .json file.
458 kwargs: Additional arguments to pass to the constructor.
460 Returns:
461 google.auth.jwt.Credentials: The constructed credentials.
462 """
463 info, signer = _service_account_info.from_filename(
464 filename, require=["client_email"]
465 )
466 return cls._from_signer_and_info(signer, info, **kwargs)
468 @classmethod
469 def from_signing_credentials(cls, credentials, audience, **kwargs):
470 """Creates a new :class:`google.auth.jwt.Credentials` instance from an
471 existing :class:`google.auth.credentials.Signing` instance.
473 The new instance will use the same signer as the existing instance and
474 will use the existing instance's signer email as the issuer and
475 subject by default.
477 Example::
479 svc_creds = service_account.Credentials.from_service_account_file(
480 'service_account.json')
481 audience = (
482 'https://pubsub.googleapis.com/google.pubsub.v1.Publisher')
483 jwt_creds = jwt.Credentials.from_signing_credentials(
484 svc_creds, audience=audience)
486 Args:
487 credentials (google.auth.credentials.Signing): The credentials to
488 use to construct the new credentials.
489 audience (str): the `aud` claim. The intended audience for the
490 credentials.
491 kwargs: Additional arguments to pass to the constructor.
493 Returns:
494 google.auth.jwt.Credentials: A new Credentials instance.
495 """
496 kwargs.setdefault("issuer", credentials.signer_email)
497 kwargs.setdefault("subject", credentials.signer_email)
498 jwt_creds = cls(credentials.signer, audience=audience, **kwargs)
500 if isinstance(
501 credentials,
502 google.auth.credentials.CredentialsWithRegionalAccessBoundary,
503 ):
504 credentials._copy_regional_access_boundary_manager(jwt_creds)
506 return jwt_creds
508 def with_claims(
509 self, issuer=None, subject=None, audience=None, additional_claims=None
510 ):
511 """Returns a copy of these credentials with modified claims.
513 Args:
514 issuer (str): The `iss` claim. If unspecified the current issuer
515 claim will be used.
516 subject (str): The `sub` claim. If unspecified the current subject
517 claim will be used.
518 audience (str): the `aud` claim. If unspecified the current
519 audience claim will be used.
520 additional_claims (Mapping[str, str]): Any additional claims for
521 the JWT payload. This will be merged with the current
522 additional claims.
524 Returns:
525 google.auth.jwt.Credentials: A new credentials instance.
526 """
527 new_additional_claims = copy.deepcopy(self._additional_claims)
528 new_additional_claims.update(additional_claims or {})
530 cred = self.__class__(
531 self._signer,
532 issuer=issuer if issuer is not None else self._issuer,
533 subject=subject if subject is not None else self._subject,
534 audience=audience if audience is not None else self._audience,
535 additional_claims=new_additional_claims,
536 quota_project_id=self._quota_project_id,
537 )
538 self._copy_regional_access_boundary_manager(cred)
539 return cred
541 @_helpers.copy_docstring(google.auth.credentials.CredentialsWithQuotaProject)
542 def with_quota_project(self, quota_project_id):
543 cred = self.__class__(
544 self._signer,
545 issuer=self._issuer,
546 subject=self._subject,
547 audience=self._audience,
548 additional_claims=self._additional_claims,
549 quota_project_id=quota_project_id,
550 )
551 self._copy_regional_access_boundary_manager(cred)
552 return cred
554 def _make_jwt(self):
555 """Make a signed JWT.
557 Returns:
558 Tuple[bytes, datetime]: The encoded JWT and the expiration.
559 """
560 now = _helpers.utcnow()
561 lifetime = datetime.timedelta(seconds=self._token_lifetime)
562 expiry = now + lifetime
564 payload = {
565 "iss": self._issuer,
566 "sub": self._subject,
567 "iat": _helpers.datetime_to_secs(now),
568 "exp": _helpers.datetime_to_secs(expiry),
569 }
570 if self._audience:
571 payload["aud"] = self._audience
573 payload.update(self._additional_claims)
575 jwt = encode(self._signer, payload)
577 return jwt, expiry
579 def _perform_refresh_token(self, request):
580 """Refreshes the access token.
582 Args:
583 request (Any): Unused.
584 """
585 # pylint: disable=unused-argument
586 # (pylint doesn't correctly recognize overridden methods.)
587 self.token, self.expiry = self._make_jwt()
589 def _build_regional_access_boundary_lookup_url(self, request=None):
590 """Builds the lookup URL using the service account's email address.
592 Returns None if the subject is populated.
593 """
594 # In jwt.Credentials, subject defaults to client_email (which is the issuer).
595 # We must check self._subject != self._issuer to correctly determine if
596 # Domain-Wide Delegation is active.
597 if self._subject and self._subject != self._issuer:
598 # RAB does not apply to Workspace User Accounts via Domain-wide Delegation.
599 return None
601 if not self.signer_email:
602 return None
604 return _regional_access_boundary_utils.get_service_account_rab_endpoint(
605 self.signer_email
606 )
608 @_helpers.copy_docstring(google.auth.credentials.Signing)
609 def sign_bytes(self, message):
610 return self._signer.sign(message)
612 @property # type: ignore
613 @_helpers.copy_docstring(google.auth.credentials.Signing)
614 def signer_email(self):
615 return self._issuer
617 @property # type: ignore
618 @_helpers.copy_docstring(google.auth.credentials.Signing)
619 def signer(self):
620 return self._signer
622 @property # type: ignore
623 def additional_claims(self):
624 """Additional claims the JWT object was created with."""
625 return self._additional_claims
628class OnDemandCredentials(
629 google.auth.credentials.Signing, google.auth.credentials.CredentialsWithQuotaProject
630):
631 """On-demand JWT credentials.
633 Like :class:`Credentials`, this class uses a JWT as the bearer token for
634 authentication. However, this class does not require the audience at
635 construction time. Instead, it will generate a new token on-demand for
636 each request using the request URI as the audience. It caches tokens
637 so that multiple requests to the same URI do not incur the overhead
638 of generating a new token every time.
640 This behavior is especially useful for `gRPC`_ clients. A gRPC service may
641 have multiple audience and gRPC clients may not know all of the audiences
642 required for accessing a particular service. With these credentials,
643 no knowledge of the audiences is required ahead of time.
645 .. _grpc: http://www.grpc.io/
646 """
648 def __init__(
649 self,
650 signer,
651 issuer,
652 subject,
653 additional_claims=None,
654 token_lifetime=_DEFAULT_TOKEN_LIFETIME_SECS,
655 max_cache_size=_DEFAULT_MAX_CACHE_SIZE,
656 quota_project_id=None,
657 ):
658 """
659 Args:
660 signer (google.auth.crypt.Signer): The signer used to sign JWTs.
661 issuer (str): The `iss` claim.
662 subject (str): The `sub` claim.
663 additional_claims (Mapping[str, str]): Any additional claims for
664 the JWT payload.
665 token_lifetime (int): The amount of time in seconds for
666 which the token is valid. Defaults to 1 hour.
667 max_cache_size (int): The maximum number of JWT tokens to keep in
668 cache. Tokens are cached using :class:`google.auth._cache.LRUCache`.
669 quota_project_id (Optional[str]): The project ID used for quota
670 and billing.
672 """
673 super(OnDemandCredentials, self).__init__()
674 self._signer = signer
675 self._issuer = issuer
676 self._subject = subject
677 self._token_lifetime = token_lifetime
678 self._quota_project_id = quota_project_id
680 if additional_claims is None:
681 additional_claims = {}
683 self._additional_claims = additional_claims
684 self._cache = _cache.LRUCache(maxsize=max_cache_size)
686 @classmethod
687 def _from_signer_and_info(cls, signer, info, **kwargs):
688 """Creates an OnDemandCredentials instance from a signer and service
689 account info.
691 Args:
692 signer (google.auth.crypt.Signer): The signer used to sign JWTs.
693 info (Mapping[str, str]): The service account info.
694 kwargs: Additional arguments to pass to the constructor.
696 Returns:
697 google.auth.jwt.OnDemandCredentials: The constructed credentials.
699 Raises:
700 google.auth.exceptions.MalformedError: If the info is not in the expected format.
701 """
702 kwargs.setdefault("subject", info["client_email"])
703 kwargs.setdefault("issuer", info["client_email"])
704 return cls(signer, **kwargs)
706 @classmethod
707 def from_service_account_info(cls, info, **kwargs):
708 """Creates an OnDemandCredentials instance from a dictionary.
710 Args:
711 info (Mapping[str, str]): The service account info in Google
712 format.
713 kwargs: Additional arguments to pass to the constructor.
715 Returns:
716 google.auth.jwt.OnDemandCredentials: The constructed credentials.
718 Raises:
719 google.auth.exceptions.MalformedError: If the info is not in the expected format.
720 """
721 signer = _service_account_info.from_dict(info, require=["client_email"])
722 return cls._from_signer_and_info(signer, info, **kwargs)
724 @classmethod
725 def from_service_account_file(cls, filename, **kwargs):
726 """Creates an OnDemandCredentials instance from a service account .json
727 file in Google format.
729 Args:
730 filename (str): The path to the service account .json file.
731 kwargs: Additional arguments to pass to the constructor.
733 Returns:
734 google.auth.jwt.OnDemandCredentials: The constructed credentials.
735 """
736 info, signer = _service_account_info.from_filename(
737 filename, require=["client_email"]
738 )
739 return cls._from_signer_and_info(signer, info, **kwargs)
741 @classmethod
742 def from_signing_credentials(cls, credentials, **kwargs):
743 """Creates a new :class:`google.auth.jwt.OnDemandCredentials` instance
744 from an existing :class:`google.auth.credentials.Signing` instance.
746 The new instance will use the same signer as the existing instance and
747 will use the existing instance's signer email as the issuer and
748 subject by default.
750 Example::
752 svc_creds = service_account.Credentials.from_service_account_file(
753 'service_account.json')
754 jwt_creds = jwt.OnDemandCredentials.from_signing_credentials(
755 svc_creds)
757 Args:
758 credentials (google.auth.credentials.Signing): The credentials to
759 use to construct the new credentials.
760 kwargs: Additional arguments to pass to the constructor.
762 Returns:
763 google.auth.jwt.Credentials: A new Credentials instance.
764 """
765 kwargs.setdefault("issuer", credentials.signer_email)
766 kwargs.setdefault("subject", credentials.signer_email)
767 return cls(credentials.signer, **kwargs)
769 def with_claims(self, issuer=None, subject=None, additional_claims=None):
770 """Returns a copy of these credentials with modified claims.
772 Args:
773 issuer (str): The `iss` claim. If unspecified the current issuer
774 claim will be used.
775 subject (str): The `sub` claim. If unspecified the current subject
776 claim will be used.
777 additional_claims (Mapping[str, str]): Any additional claims for
778 the JWT payload. This will be merged with the current
779 additional claims.
781 Returns:
782 google.auth.jwt.OnDemandCredentials: A new credentials instance.
783 """
784 new_additional_claims = copy.deepcopy(self._additional_claims)
785 new_additional_claims.update(additional_claims or {})
787 return self.__class__(
788 self._signer,
789 issuer=issuer if issuer is not None else self._issuer,
790 subject=subject if subject is not None else self._subject,
791 additional_claims=new_additional_claims,
792 max_cache_size=self._cache.maxsize,
793 quota_project_id=self._quota_project_id,
794 )
796 @_helpers.copy_docstring(google.auth.credentials.CredentialsWithQuotaProject)
797 def with_quota_project(self, quota_project_id):
798 return self.__class__(
799 self._signer,
800 issuer=self._issuer,
801 subject=self._subject,
802 additional_claims=self._additional_claims,
803 max_cache_size=self._cache.maxsize,
804 quota_project_id=quota_project_id,
805 )
807 @property
808 def valid(self):
809 """Checks the validity of the credentials.
811 These credentials are always valid because it generates tokens on
812 demand.
813 """
814 return True
816 def _make_jwt_for_audience(self, audience):
817 """Make a new JWT for the given audience.
819 Args:
820 audience (str): The intended audience.
822 Returns:
823 Tuple[bytes, datetime]: The encoded JWT and the expiration.
824 """
825 now = _helpers.utcnow()
826 lifetime = datetime.timedelta(seconds=self._token_lifetime)
827 expiry = now + lifetime
829 payload = {
830 "iss": self._issuer,
831 "sub": self._subject,
832 "iat": _helpers.datetime_to_secs(now),
833 "exp": _helpers.datetime_to_secs(expiry),
834 "aud": audience,
835 }
837 payload.update(self._additional_claims)
839 jwt = encode(self._signer, payload)
841 return jwt, expiry
843 def _get_jwt_for_audience(self, audience):
844 """Get a JWT For a given audience.
846 If there is already an existing, non-expired token in the cache for
847 the audience, that token is used. Otherwise, a new token will be
848 created.
850 Args:
851 audience (str): The intended audience.
853 Returns:
854 bytes: The encoded JWT.
855 """
856 token, expiry = self._cache.get(audience, (None, None))
858 if token is None or expiry < _helpers.utcnow():
859 token, expiry = self._make_jwt_for_audience(audience)
860 self._cache[audience] = token, expiry
862 return token
864 def refresh(self, request):
865 """Raises an exception, these credentials can not be directly
866 refreshed.
868 Args:
869 request (Any): Unused.
871 Raises:
872 google.auth.RefreshError
873 """
874 # pylint: disable=unused-argument
875 # (pylint doesn't correctly recognize overridden methods.)
876 raise exceptions.RefreshError(
877 "OnDemandCredentials can not be directly refreshed."
878 )
880 def before_request(self, request, method, url, headers):
881 """Performs credential-specific before request logic.
883 Args:
884 request (Any): Unused. JWT credentials do not need to make an
885 HTTP request to refresh.
886 method (str): The request's HTTP method.
887 url (str): The request's URI. This is used as the audience claim
888 when generating the JWT.
889 headers (Mapping): The request's headers.
890 """
891 # pylint: disable=unused-argument
892 # (pylint doesn't correctly recognize overridden methods.)
893 parts = urllib.parse.urlsplit(url)
894 # Strip query string and fragment
895 audience = urllib.parse.urlunsplit(
896 (parts.scheme, parts.netloc, parts.path, "", "")
897 )
898 token = self._get_jwt_for_audience(audience)
899 self.apply(headers, token=token)
901 @_helpers.copy_docstring(google.auth.credentials.Signing)
902 def sign_bytes(self, message):
903 return self._signer.sign(message)
905 @property # type: ignore
906 @_helpers.copy_docstring(google.auth.credentials.Signing)
907 def signer_email(self):
908 return self._issuer
910 @property # type: ignore
911 @_helpers.copy_docstring(google.auth.credentials.Signing)
912 def signer(self):
913 return self._signer