Source code for google.auth.jwt

# Copyright 2016 Google LLC
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#      http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.

"""JSON Web Tokens

Provides support for creating (encoding) and verifying (decoding) JWTs,
especially JWTs generated and consumed by Google infrastructure.

See `rfc7519`_ for more details on JWTs.

To encode a JWT use :func:`encode`::

    from google.auth import crypt
    from google.auth import jwt

    signer = crypt.Signer(private_key)
    payload = {'some': 'payload'}
    encoded = jwt.encode(signer, payload)

To decode a JWT and verify claims use :func:`decode`::

    claims = jwt.decode(encoded, certs=public_certs)

You can also skip verification::

    claims = jwt.decode(encoded, verify=False)

.. _rfc7519: https://tools.ietf.org/html/rfc7519

"""

try:
    from collections.abc import Mapping
# Python 2.7 compatibility
except ImportError:  # pragma: NO COVER
    from collections import Mapping
import copy
import datetime
import json

import cachetools
import six
from six.moves import urllib

from google.auth import _helpers
from google.auth import _service_account_info
from google.auth import crypt
from google.auth import exceptions
import google.auth.credentials

try:
    from google.auth.crypt import es256
except ImportError:  # pragma: NO COVER
    es256 = None

_DEFAULT_TOKEN_LIFETIME_SECS = 3600  # 1 hour in seconds
_DEFAULT_MAX_CACHE_SIZE = 10
_ALGORITHM_TO_VERIFIER_CLASS = {"RS256": crypt.RSAVerifier}
_CRYPTOGRAPHY_BASED_ALGORITHMS = set(["ES256"])

if es256 is not None:  # pragma: NO COVER
    _ALGORITHM_TO_VERIFIER_CLASS["ES256"] = es256.ES256Verifier


[docs]def encode(signer, payload, header=None, key_id=None): """Make a signed JWT. Args: signer (google.auth.crypt.Signer): The signer used to sign the JWT. payload (Mapping[str, str]): The JWT payload. header (Mapping[str, str]): Additional JWT header payload. key_id (str): The key id to add to the JWT header. If the signer has a key id it will be used as the default. If this is specified it will override the signer's key id. Returns: bytes: The encoded JWT. """ if header is None: header = {} if key_id is None: key_id = signer.key_id header.update({"typ": "JWT"}) if es256 is not None and isinstance(signer, es256.ES256Signer): header.update({"alg": "ES256"}) else: header.update({"alg": "RS256"}) if key_id is not None: header["kid"] = key_id segments = [ _helpers.unpadded_urlsafe_b64encode(json.dumps(header).encode("utf-8")), _helpers.unpadded_urlsafe_b64encode(json.dumps(payload).encode("utf-8")), ] signing_input = b".".join(segments) signature = signer.sign(signing_input) segments.append(_helpers.unpadded_urlsafe_b64encode(signature)) return b".".join(segments)
def _decode_jwt_segment(encoded_section): """Decodes a single JWT segment.""" section_bytes = _helpers.padded_urlsafe_b64decode(encoded_section) try: return json.loads(section_bytes.decode("utf-8")) except ValueError as caught_exc: new_exc = ValueError("Can't parse segment: {0}".format(section_bytes)) six.raise_from(new_exc, caught_exc) def _unverified_decode(token): """Decodes a token and does no verification. Args: token (Union[str, bytes]): The encoded JWT. Returns: Tuple[str, str, str, str]: header, payload, signed_section, and signature. Raises: ValueError: if there are an incorrect amount of segments in the token. """ token = _helpers.to_bytes(token) if token.count(b".") != 2: raise ValueError("Wrong number of segments in token: {0}".format(token)) encoded_header, encoded_payload, signature = token.split(b".") signed_section = encoded_header + b"." + encoded_payload signature = _helpers.padded_urlsafe_b64decode(signature) # Parse segments header = _decode_jwt_segment(encoded_header) payload = _decode_jwt_segment(encoded_payload) return header, payload, signed_section, signature
[docs]def decode_header(token): """Return the decoded header of a token. No verification is done. This is useful to extract the key id from the header in order to acquire the appropriate certificate to verify the token. Args: token (Union[str, bytes]): the encoded JWT. Returns: Mapping: The decoded JWT header. """ header, _, _, _ = _unverified_decode(token) return header
def _verify_iat_and_exp(payload): """Verifies the ``iat`` (Issued At) and ``exp`` (Expires) claims in a token payload. Args: payload (Mapping[str, str]): The JWT payload. Raises: ValueError: if any checks failed. """ now = _helpers.datetime_to_secs(_helpers.utcnow()) # Make sure the iat and exp claims are present. for key in ("iat", "exp"): if key not in payload: raise ValueError("Token does not contain required claim {}".format(key)) # Make sure the token wasn't issued in the future. iat = payload["iat"] # Err on the side of accepting a token that is slightly early to account # for clock skew. earliest = iat - _helpers.CLOCK_SKEW_SECS if now < earliest: raise ValueError("Token used too early, {} < {}".format(now, iat)) # Make sure the token wasn't issued in the past. exp = payload["exp"] # Err on the side of accepting a token that is slightly out of date # to account for clow skew. latest = exp + _helpers.CLOCK_SKEW_SECS if latest < now: raise ValueError("Token expired, {} < {}".format(latest, now))
[docs]def decode(token, certs=None, verify=True, audience=None): """Decode and verify a JWT. Args: token (str): The encoded JWT. certs (Union[str, bytes, Mapping[str, Union[str, bytes]]]): The certificate used to validate the JWT signature. If bytes or string, it must the the public key certificate in PEM format. If a mapping, it must be a mapping of key IDs to public key certificates in PEM format. The mapping must contain the same key ID that's specified in the token's header. verify (bool): Whether to perform signature and claim validation. Verification is done by default. audience (str): The audience claim, 'aud', that this JWT should contain. If None then the JWT's 'aud' parameter is not verified. Returns: Mapping[str, str]: The deserialized JSON payload in the JWT. Raises: ValueError: if any verification checks failed. """ header, payload, signed_section, signature = _unverified_decode(token) if not verify: return payload # Pluck the key id and algorithm from the header and make sure we have # a verifier that can support it. key_alg = header.get("alg") key_id = header.get("kid") try: verifier_cls = _ALGORITHM_TO_VERIFIER_CLASS[key_alg] except KeyError as exc: if key_alg in _CRYPTOGRAPHY_BASED_ALGORITHMS: six.raise_from( ValueError( "The key algorithm {} requires the cryptography package " "to be installed.".format(key_alg) ), exc, ) else: six.raise_from( ValueError("Unsupported signature algorithm {}".format(key_alg)), exc ) # If certs is specified as a dictionary of key IDs to certificates, then # use the certificate identified by the key ID in the token header. if isinstance(certs, Mapping): if key_id: if key_id not in certs: raise ValueError("Certificate for key id {} not found.".format(key_id)) certs_to_check = [certs[key_id]] # If there's no key id in the header, check against all of the certs. else: certs_to_check = certs.values() else: certs_to_check = certs # Verify that the signature matches the message. if not crypt.verify_signature( signed_section, signature, certs_to_check, verifier_cls ): raise ValueError("Could not verify token signature.") # Verify the issued at and created times in the payload. _verify_iat_and_exp(payload) # Check audience. if audience is not None: claim_audience = payload.get("aud") if audience != claim_audience: raise ValueError( "Token has wrong audience {}, expected {}".format( claim_audience, audience ) ) return payload
[docs]class Credentials(google.auth.credentials.Signing, google.auth.credentials.Credentials): """Credentials that use a JWT as the bearer token. These credentials require an "audience" claim. This claim identifies the intended recipient of the bearer token. The constructor arguments determine the claims for the JWT that is sent with requests. Usually, you'll construct these credentials with one of the helper constructors as shown in the next section. To create JWT credentials using a Google service account private key JSON file:: audience = 'https://pubsub.googleapis.com/google.pubsub.v1.Publisher' credentials = jwt.Credentials.from_service_account_file( 'service-account.json', audience=audience) If you already have the service account file loaded and parsed:: service_account_info = json.load(open('service_account.json')) credentials = jwt.Credentials.from_service_account_info( service_account_info, audience=audience) Both helper methods pass on arguments to the constructor, so you can specify the JWT claims:: credentials = jwt.Credentials.from_service_account_file( 'service-account.json', audience=audience, additional_claims={'meta': 'data'}) You can also construct the credentials directly if you have a :class:`~google.auth.crypt.Signer` instance:: credentials = jwt.Credentials( signer, issuer='your-issuer', subject='your-subject', audience=audience) The claims are considered immutable. If you want to modify the claims, you can easily create another instance using :meth:`with_claims`:: new_audience = ( 'https://pubsub.googleapis.com/google.pubsub.v1.Subscriber') new_credentials = credentials.with_claims(audience=new_audience) """ def __init__( self, signer, issuer, subject, audience, additional_claims=None, token_lifetime=_DEFAULT_TOKEN_LIFETIME_SECS, ): """ Args: signer (google.auth.crypt.Signer): The signer used to sign JWTs. issuer (str): The `iss` claim. subject (str): The `sub` claim. audience (str): the `aud` claim. The intended audience for the credentials. additional_claims (Mapping[str, str]): Any additional claims for the JWT payload. token_lifetime (int): The amount of time in seconds for which the token is valid. Defaults to 1 hour. """ super(Credentials, self).__init__() self._signer = signer self._issuer = issuer self._subject = subject self._audience = audience self._token_lifetime = token_lifetime if additional_claims is None: additional_claims = {} self._additional_claims = additional_claims @classmethod def _from_signer_and_info(cls, signer, info, **kwargs): """Creates a Credentials instance from a signer and service account info. Args: signer (google.auth.crypt.Signer): The signer used to sign JWTs. info (Mapping[str, str]): The service account info. kwargs: Additional arguments to pass to the constructor. Returns: google.auth.jwt.Credentials: The constructed credentials. Raises: ValueError: If the info is not in the expected format. """ kwargs.setdefault("subject", info["client_email"]) kwargs.setdefault("issuer", info["client_email"]) return cls(signer, **kwargs)
[docs] @classmethod def from_service_account_info(cls, info, **kwargs): """Creates an Credentials instance from a dictionary. Args: info (Mapping[str, str]): The service account info in Google format. kwargs: Additional arguments to pass to the constructor. Returns: google.auth.jwt.Credentials: The constructed credentials. Raises: ValueError: If the info is not in the expected format. """ signer = _service_account_info.from_dict(info, require=["client_email"]) return cls._from_signer_and_info(signer, info, **kwargs)
[docs] @classmethod def from_service_account_file(cls, filename, **kwargs): """Creates a Credentials instance from a service account .json file in Google format. Args: filename (str): The path to the service account .json file. kwargs: Additional arguments to pass to the constructor. Returns: google.auth.jwt.Credentials: The constructed credentials. """ info, signer = _service_account_info.from_filename( filename, require=["client_email"] ) return cls._from_signer_and_info(signer, info, **kwargs)
[docs] @classmethod def from_signing_credentials(cls, credentials, audience, **kwargs): """Creates a new :class:`google.auth.jwt.Credentials` instance from an existing :class:`google.auth.credentials.Signing` instance. The new instance will use the same signer as the existing instance and will use the existing instance's signer email as the issuer and subject by default. Example:: svc_creds = service_account.Credentials.from_service_account_file( 'service_account.json') audience = ( 'https://pubsub.googleapis.com/google.pubsub.v1.Publisher') jwt_creds = jwt.Credentials.from_signing_credentials( svc_creds, audience=audience) Args: credentials (google.auth.credentials.Signing): The credentials to use to construct the new credentials. audience (str): the `aud` claim. The intended audience for the credentials. kwargs: Additional arguments to pass to the constructor. Returns: google.auth.jwt.Credentials: A new Credentials instance. """ kwargs.setdefault("issuer", credentials.signer_email) kwargs.setdefault("subject", credentials.signer_email) return cls(credentials.signer, audience=audience, **kwargs)
[docs] def with_claims( self, issuer=None, subject=None, audience=None, additional_claims=None ): """Returns a copy of these credentials with modified claims. Args: issuer (str): The `iss` claim. If unspecified the current issuer claim will be used. subject (str): The `sub` claim. If unspecified the current subject claim will be used. audience (str): the `aud` claim. If unspecified the current audience claim will be used. additional_claims (Mapping[str, str]): Any additional claims for the JWT payload. This will be merged with the current additional claims. Returns: google.auth.jwt.Credentials: A new credentials instance. """ new_additional_claims = copy.deepcopy(self._additional_claims) new_additional_claims.update(additional_claims or {}) return self.__class__( self._signer, issuer=issuer if issuer is not None else self._issuer, subject=subject if subject is not None else self._subject, audience=audience if audience is not None else self._audience, additional_claims=new_additional_claims, )
def _make_jwt(self): """Make a signed JWT. Returns: Tuple[bytes, datetime]: The encoded JWT and the expiration. """ now = _helpers.utcnow() lifetime = datetime.timedelta(seconds=self._token_lifetime) expiry = now + lifetime payload = { "iss": self._issuer, "sub": self._subject, "iat": _helpers.datetime_to_secs(now), "exp": _helpers.datetime_to_secs(expiry), "aud": self._audience, } payload.update(self._additional_claims) jwt = encode(self._signer, payload) return jwt, expiry
[docs] def refresh(self, request): """Refreshes the access token. Args: request (Any): Unused. """ # pylint: disable=unused-argument # (pylint doesn't correctly recognize overridden methods.) self.token, self.expiry = self._make_jwt()
[docs] @_helpers.copy_docstring(google.auth.credentials.Signing) def sign_bytes(self, message): return self._signer.sign(message)
@property @_helpers.copy_docstring(google.auth.credentials.Signing) def signer_email(self): return self._issuer @property @_helpers.copy_docstring(google.auth.credentials.Signing) def signer(self): return self._signer
[docs]class OnDemandCredentials( google.auth.credentials.Signing, google.auth.credentials.Credentials ): """On-demand JWT credentials. Like :class:`Credentials`, this class uses a JWT as the bearer token for authentication. However, this class does not require the audience at construction time. Instead, it will generate a new token on-demand for each request using the request URI as the audience. It caches tokens so that multiple requests to the same URI do not incur the overhead of generating a new token every time. This behavior is especially useful for `gRPC`_ clients. A gRPC service may have multiple audience and gRPC clients may not know all of the audiences required for accessing a particular service. With these credentials, no knowledge of the audiences is required ahead of time. .. _grpc: http://www.grpc.io/ """ def __init__( self, signer, issuer, subject, additional_claims=None, token_lifetime=_DEFAULT_TOKEN_LIFETIME_SECS, max_cache_size=_DEFAULT_MAX_CACHE_SIZE, ): """ Args: signer (google.auth.crypt.Signer): The signer used to sign JWTs. issuer (str): The `iss` claim. subject (str): The `sub` claim. additional_claims (Mapping[str, str]): Any additional claims for the JWT payload. token_lifetime (int): The amount of time in seconds for which the token is valid. Defaults to 1 hour. max_cache_size (int): The maximum number of JWT tokens to keep in cache. Tokens are cached using :class:`cachetools.LRUCache`. """ super(OnDemandCredentials, self).__init__() self._signer = signer self._issuer = issuer self._subject = subject self._token_lifetime = token_lifetime if additional_claims is None: additional_claims = {} self._additional_claims = additional_claims self._cache = cachetools.LRUCache(maxsize=max_cache_size) @classmethod def _from_signer_and_info(cls, signer, info, **kwargs): """Creates an OnDemandCredentials instance from a signer and service account info. Args: signer (google.auth.crypt.Signer): The signer used to sign JWTs. info (Mapping[str, str]): The service account info. kwargs: Additional arguments to pass to the constructor. Returns: google.auth.jwt.OnDemandCredentials: The constructed credentials. Raises: ValueError: If the info is not in the expected format. """ kwargs.setdefault("subject", info["client_email"]) kwargs.setdefault("issuer", info["client_email"]) return cls(signer, **kwargs)
[docs] @classmethod def from_service_account_info(cls, info, **kwargs): """Creates an OnDemandCredentials instance from a dictionary. Args: info (Mapping[str, str]): The service account info in Google format. kwargs: Additional arguments to pass to the constructor. Returns: google.auth.jwt.OnDemandCredentials: The constructed credentials. Raises: ValueError: If the info is not in the expected format. """ signer = _service_account_info.from_dict(info, require=["client_email"]) return cls._from_signer_and_info(signer, info, **kwargs)
[docs] @classmethod def from_service_account_file(cls, filename, **kwargs): """Creates an OnDemandCredentials instance from a service account .json file in Google format. Args: filename (str): The path to the service account .json file. kwargs: Additional arguments to pass to the constructor. Returns: google.auth.jwt.OnDemandCredentials: The constructed credentials. """ info, signer = _service_account_info.from_filename( filename, require=["client_email"] ) return cls._from_signer_and_info(signer, info, **kwargs)
[docs] @classmethod def from_signing_credentials(cls, credentials, **kwargs): """Creates a new :class:`google.auth.jwt.OnDemandCredentials` instance from an existing :class:`google.auth.credentials.Signing` instance. The new instance will use the same signer as the existing instance and will use the existing instance's signer email as the issuer and subject by default. Example:: svc_creds = service_account.Credentials.from_service_account_file( 'service_account.json') jwt_creds = jwt.OnDemandCredentials.from_signing_credentials( svc_creds) Args: credentials (google.auth.credentials.Signing): The credentials to use to construct the new credentials. kwargs: Additional arguments to pass to the constructor. Returns: google.auth.jwt.Credentials: A new Credentials instance. """ kwargs.setdefault("issuer", credentials.signer_email) kwargs.setdefault("subject", credentials.signer_email) return cls(credentials.signer, **kwargs)
[docs] def with_claims(self, issuer=None, subject=None, additional_claims=None): """Returns a copy of these credentials with modified claims. Args: issuer (str): The `iss` claim. If unspecified the current issuer claim will be used. subject (str): The `sub` claim. If unspecified the current subject claim will be used. additional_claims (Mapping[str, str]): Any additional claims for the JWT payload. This will be merged with the current additional claims. Returns: google.auth.jwt.OnDemandCredentials: A new credentials instance. """ new_additional_claims = copy.deepcopy(self._additional_claims) new_additional_claims.update(additional_claims or {}) return self.__class__( self._signer, issuer=issuer if issuer is not None else self._issuer, subject=subject if subject is not None else self._subject, additional_claims=new_additional_claims, max_cache_size=self._cache.maxsize, )
@property def valid(self): """Checks the validity of the credentials. These credentials are always valid because it generates tokens on demand. """ return True def _make_jwt_for_audience(self, audience): """Make a new JWT for the given audience. Args: audience (str): The intended audience. Returns: Tuple[bytes, datetime]: The encoded JWT and the expiration. """ now = _helpers.utcnow() lifetime = datetime.timedelta(seconds=self._token_lifetime) expiry = now + lifetime payload = { "iss": self._issuer, "sub": self._subject, "iat": _helpers.datetime_to_secs(now), "exp": _helpers.datetime_to_secs(expiry), "aud": audience, } payload.update(self._additional_claims) jwt = encode(self._signer, payload) return jwt, expiry def _get_jwt_for_audience(self, audience): """Get a JWT For a given audience. If there is already an existing, non-expired token in the cache for the audience, that token is used. Otherwise, a new token will be created. Args: audience (str): The intended audience. Returns: bytes: The encoded JWT. """ token, expiry = self._cache.get(audience, (None, None)) if token is None or expiry < _helpers.utcnow(): token, expiry = self._make_jwt_for_audience(audience) self._cache[audience] = token, expiry return token
[docs] def refresh(self, request): """Raises an exception, these credentials can not be directly refreshed. Args: request (Any): Unused. Raises: google.auth.RefreshError """ # pylint: disable=unused-argument # (pylint doesn't correctly recognize overridden methods.) raise exceptions.RefreshError( "OnDemandCredentials can not be directly refreshed." )
[docs] def before_request(self, request, method, url, headers): """Performs credential-specific before request logic. Args: request (Any): Unused. JWT credentials do not need to make an HTTP request to refresh. method (str): The request's HTTP method. url (str): The request's URI. This is used as the audience claim when generating the JWT. headers (Mapping): The request's headers. """ # pylint: disable=unused-argument # (pylint doesn't correctly recognize overridden methods.) parts = urllib.parse.urlsplit(url) # Strip query string and fragment audience = urllib.parse.urlunsplit( (parts.scheme, parts.netloc, parts.path, "", "") ) token = self._get_jwt_for_audience(audience) self.apply(headers, token=token)
[docs] @_helpers.copy_docstring(google.auth.credentials.Signing) def sign_bytes(self, message): return self._signer.sign(message)
@property @_helpers.copy_docstring(google.auth.credentials.Signing) def signer_email(self): return self._issuer @property @_helpers.copy_docstring(google.auth.credentials.Signing) def signer(self): return self._signer