| 123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250 |
- # -*- coding: utf-8 -*-
- #
- # Copyright 2014 Google Inc. All rights reserved.
- #
- # 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.
- """Crypto-related routines for oauth2client."""
- import json
- import logging
- import time
- from oauth2client import _helpers
- from oauth2client import _pure_python_crypt
- RsaSigner = _pure_python_crypt.RsaSigner
- RsaVerifier = _pure_python_crypt.RsaVerifier
- CLOCK_SKEW_SECS = 300 # 5 minutes in seconds
- AUTH_TOKEN_LIFETIME_SECS = 300 # 5 minutes in seconds
- MAX_TOKEN_LIFETIME_SECS = 86400 # 1 day in seconds
- logger = logging.getLogger(__name__)
- class AppIdentityError(Exception):
- """Error to indicate crypto failure."""
- def _bad_pkcs12_key_as_pem(*args, **kwargs):
- raise NotImplementedError('pkcs12_key_as_pem requires OpenSSL.')
- try:
- from oauth2client import _openssl_crypt
- OpenSSLSigner = _openssl_crypt.OpenSSLSigner
- OpenSSLVerifier = _openssl_crypt.OpenSSLVerifier
- pkcs12_key_as_pem = _openssl_crypt.pkcs12_key_as_pem
- except ImportError: # pragma: NO COVER
- OpenSSLVerifier = None
- OpenSSLSigner = None
- pkcs12_key_as_pem = _bad_pkcs12_key_as_pem
- try:
- from oauth2client import _pycrypto_crypt
- PyCryptoSigner = _pycrypto_crypt.PyCryptoSigner
- PyCryptoVerifier = _pycrypto_crypt.PyCryptoVerifier
- except ImportError: # pragma: NO COVER
- PyCryptoVerifier = None
- PyCryptoSigner = None
- if OpenSSLSigner:
- Signer = OpenSSLSigner
- Verifier = OpenSSLVerifier
- elif PyCryptoSigner: # pragma: NO COVER
- Signer = PyCryptoSigner
- Verifier = PyCryptoVerifier
- else: # pragma: NO COVER
- Signer = RsaSigner
- Verifier = RsaVerifier
- def make_signed_jwt(signer, payload, key_id=None):
- """Make a signed JWT.
- See http://self-issued.info/docs/draft-jones-json-web-token.html.
- Args:
- signer: crypt.Signer, Cryptographic signer.
- payload: dict, Dictionary of data to convert to JSON and then sign.
- key_id: string, (Optional) Key ID header.
- Returns:
- string, The JWT for the payload.
- """
- header = {'typ': 'JWT', 'alg': 'RS256'}
- if key_id is not None:
- header['kid'] = key_id
- segments = [
- _helpers._urlsafe_b64encode(_helpers._json_encode(header)),
- _helpers._urlsafe_b64encode(_helpers._json_encode(payload)),
- ]
- signing_input = b'.'.join(segments)
- signature = signer.sign(signing_input)
- segments.append(_helpers._urlsafe_b64encode(signature))
- logger.debug(str(segments))
- return b'.'.join(segments)
- def _verify_signature(message, signature, certs):
- """Verifies signed content using a list of certificates.
- Args:
- message: string or bytes, The message to verify.
- signature: string or bytes, The signature on the message.
- certs: iterable, certificates in PEM format.
- Raises:
- AppIdentityError: If none of the certificates can verify the message
- against the signature.
- """
- for pem in certs:
- verifier = Verifier.from_string(pem, is_x509_cert=True)
- if verifier.verify(message, signature):
- return
- # If we have not returned, no certificate confirms the signature.
- raise AppIdentityError('Invalid token signature')
- def _check_audience(payload_dict, audience):
- """Checks audience field from a JWT payload.
- Does nothing if the passed in ``audience`` is null.
- Args:
- payload_dict: dict, A dictionary containing a JWT payload.
- audience: string or NoneType, an audience to check for in
- the JWT payload.
- Raises:
- AppIdentityError: If there is no ``'aud'`` field in the payload
- dictionary but there is an ``audience`` to check.
- AppIdentityError: If the ``'aud'`` field in the payload dictionary
- does not match the ``audience``.
- """
- if audience is None:
- return
- audience_in_payload = payload_dict.get('aud')
- if audience_in_payload is None:
- raise AppIdentityError(
- 'No aud field in token: {0}'.format(payload_dict))
- if audience_in_payload != audience:
- raise AppIdentityError('Wrong recipient, {0} != {1}: {2}'.format(
- audience_in_payload, audience, payload_dict))
- def _verify_time_range(payload_dict):
- """Verifies the issued at and expiration from a JWT payload.
- Makes sure the current time (in UTC) falls between the issued at and
- expiration for the JWT (with some skew allowed for via
- ``CLOCK_SKEW_SECS``).
- Args:
- payload_dict: dict, A dictionary containing a JWT payload.
- Raises:
- AppIdentityError: If there is no ``'iat'`` field in the payload
- dictionary.
- AppIdentityError: If there is no ``'exp'`` field in the payload
- dictionary.
- AppIdentityError: If the JWT expiration is too far in the future (i.e.
- if the expiration would imply a token lifetime
- longer than what is allowed.)
- AppIdentityError: If the token appears to have been issued in the
- future (up to clock skew).
- AppIdentityError: If the token appears to have expired in the past
- (up to clock skew).
- """
- # Get the current time to use throughout.
- now = int(time.time())
- # Make sure issued at and expiration are in the payload.
- issued_at = payload_dict.get('iat')
- if issued_at is None:
- raise AppIdentityError(
- 'No iat field in token: {0}'.format(payload_dict))
- expiration = payload_dict.get('exp')
- if expiration is None:
- raise AppIdentityError(
- 'No exp field in token: {0}'.format(payload_dict))
- # Make sure the expiration gives an acceptable token lifetime.
- if expiration >= now + MAX_TOKEN_LIFETIME_SECS:
- raise AppIdentityError(
- 'exp field too far in future: {0}'.format(payload_dict))
- # Make sure (up to clock skew) that the token wasn't issued in the future.
- earliest = issued_at - CLOCK_SKEW_SECS
- if now < earliest:
- raise AppIdentityError('Token used too early, {0} < {1}: {2}'.format(
- now, earliest, payload_dict))
- # Make sure (up to clock skew) that the token isn't already expired.
- latest = expiration + CLOCK_SKEW_SECS
- if now > latest:
- raise AppIdentityError('Token used too late, {0} > {1}: {2}'.format(
- now, latest, payload_dict))
- def verify_signed_jwt_with_certs(jwt, certs, audience=None):
- """Verify a JWT against public certs.
- See http://self-issued.info/docs/draft-jones-json-web-token.html.
- Args:
- jwt: string, A JWT.
- certs: dict, Dictionary where values of public keys in PEM format.
- audience: string, The audience, 'aud', that this JWT should contain. If
- None then the JWT's 'aud' parameter is not verified.
- Returns:
- dict, The deserialized JSON payload in the JWT.
- Raises:
- AppIdentityError: if any checks are failed.
- """
- jwt = _helpers._to_bytes(jwt)
- if jwt.count(b'.') != 2:
- raise AppIdentityError(
- 'Wrong number of segments in token: {0}'.format(jwt))
- header, payload, signature = jwt.split(b'.')
- message_to_sign = header + b'.' + payload
- signature = _helpers._urlsafe_b64decode(signature)
- # Parse token.
- payload_bytes = _helpers._urlsafe_b64decode(payload)
- try:
- payload_dict = json.loads(_helpers._from_bytes(payload_bytes))
- except:
- raise AppIdentityError('Can\'t parse token: {0}'.format(payload_bytes))
- # Verify that the signature matches the message.
- _verify_signature(message_to_sign, signature, certs.values())
- # Verify the issued at and created times in the payload.
- _verify_time_range(payload_dict)
- # Check audience.
- _check_audience(payload_dict, audience)
- return payload_dict
|