Index: reviewbot/third_party/google-api-python-client/oauth2client/crypt.py |
=================================================================== |
--- reviewbot/third_party/google-api-python-client/oauth2client/crypt.py (revision 221173) |
+++ reviewbot/third_party/google-api-python-client/oauth2client/crypt.py (working copy) |
@@ -1,377 +0,0 @@ |
-#!/usr/bin/python2.4 |
-# -*- coding: utf-8 -*- |
-# |
-# Copyright (C) 2011 Google Inc. |
-# |
-# 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. |
- |
-import base64 |
-import hashlib |
-import logging |
-import time |
- |
-from anyjson import simplejson |
- |
- |
-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): |
- pass |
- |
- |
-try: |
- from OpenSSL import crypto |
- |
- |
- class OpenSSLVerifier(object): |
- """Verifies the signature on a message.""" |
- |
- def __init__(self, pubkey): |
- """Constructor. |
- |
- Args: |
- pubkey, OpenSSL.crypto.PKey, The public key to verify with. |
- """ |
- self._pubkey = pubkey |
- |
- def verify(self, message, signature): |
- """Verifies a message against a signature. |
- |
- Args: |
- message: string, The message to verify. |
- signature: string, The signature on the message. |
- |
- Returns: |
- True if message was signed by the private key associated with the public |
- key that this object was constructed with. |
- """ |
- try: |
- crypto.verify(self._pubkey, signature, message, 'sha256') |
- return True |
- except: |
- return False |
- |
- @staticmethod |
- def from_string(key_pem, is_x509_cert): |
- """Construct a Verified instance from a string. |
- |
- Args: |
- key_pem: string, public key in PEM format. |
- is_x509_cert: bool, True if key_pem is an X509 cert, otherwise it is |
- expected to be an RSA key in PEM format. |
- |
- Returns: |
- Verifier instance. |
- |
- Raises: |
- OpenSSL.crypto.Error if the key_pem can't be parsed. |
- """ |
- if is_x509_cert: |
- pubkey = crypto.load_certificate(crypto.FILETYPE_PEM, key_pem) |
- else: |
- pubkey = crypto.load_privatekey(crypto.FILETYPE_PEM, key_pem) |
- return OpenSSLVerifier(pubkey) |
- |
- |
- class OpenSSLSigner(object): |
- """Signs messages with a private key.""" |
- |
- def __init__(self, pkey): |
- """Constructor. |
- |
- Args: |
- pkey, OpenSSL.crypto.PKey (or equiv), The private key to sign with. |
- """ |
- self._key = pkey |
- |
- def sign(self, message): |
- """Signs a message. |
- |
- Args: |
- message: string, Message to be signed. |
- |
- Returns: |
- string, The signature of the message for the given key. |
- """ |
- return crypto.sign(self._key, message, 'sha256') |
- |
- @staticmethod |
- def from_string(key, password='notasecret'): |
- """Construct a Signer instance from a string. |
- |
- Args: |
- key: string, private key in PKCS12 or PEM format. |
- password: string, password for the private key file. |
- |
- Returns: |
- Signer instance. |
- |
- Raises: |
- OpenSSL.crypto.Error if the key can't be parsed. |
- """ |
- if key.startswith('-----BEGIN '): |
- pkey = crypto.load_privatekey(crypto.FILETYPE_PEM, key) |
- else: |
- pkey = crypto.load_pkcs12(key, password).get_privatekey() |
- return OpenSSLSigner(pkey) |
- |
-except ImportError: |
- OpenSSLVerifier = None |
- OpenSSLSigner = None |
- |
- |
-try: |
- from Crypto.PublicKey import RSA |
- from Crypto.Hash import SHA256 |
- from Crypto.Signature import PKCS1_v1_5 |
- |
- |
- class PyCryptoVerifier(object): |
- """Verifies the signature on a message.""" |
- |
- def __init__(self, pubkey): |
- """Constructor. |
- |
- Args: |
- pubkey, OpenSSL.crypto.PKey (or equiv), The public key to verify with. |
- """ |
- self._pubkey = pubkey |
- |
- def verify(self, message, signature): |
- """Verifies a message against a signature. |
- |
- Args: |
- message: string, The message to verify. |
- signature: string, The signature on the message. |
- |
- Returns: |
- True if message was signed by the private key associated with the public |
- key that this object was constructed with. |
- """ |
- try: |
- return PKCS1_v1_5.new(self._pubkey).verify( |
- SHA256.new(message), signature) |
- except: |
- return False |
- |
- @staticmethod |
- def from_string(key_pem, is_x509_cert): |
- """Construct a Verified instance from a string. |
- |
- Args: |
- key_pem: string, public key in PEM format. |
- is_x509_cert: bool, True if key_pem is an X509 cert, otherwise it is |
- expected to be an RSA key in PEM format. |
- |
- Returns: |
- Verifier instance. |
- |
- Raises: |
- NotImplementedError if is_x509_cert is true. |
- """ |
- if is_x509_cert: |
- raise NotImplementedError( |
- 'X509 certs are not supported by the PyCrypto library. ' |
- 'Try using PyOpenSSL if native code is an option.') |
- else: |
- pubkey = RSA.importKey(key_pem) |
- return PyCryptoVerifier(pubkey) |
- |
- |
- class PyCryptoSigner(object): |
- """Signs messages with a private key.""" |
- |
- def __init__(self, pkey): |
- """Constructor. |
- |
- Args: |
- pkey, OpenSSL.crypto.PKey (or equiv), The private key to sign with. |
- """ |
- self._key = pkey |
- |
- def sign(self, message): |
- """Signs a message. |
- |
- Args: |
- message: string, Message to be signed. |
- |
- Returns: |
- string, The signature of the message for the given key. |
- """ |
- return PKCS1_v1_5.new(self._key).sign(SHA256.new(message)) |
- |
- @staticmethod |
- def from_string(key, password='notasecret'): |
- """Construct a Signer instance from a string. |
- |
- Args: |
- key: string, private key in PEM format. |
- password: string, password for private key file. Unused for PEM files. |
- |
- Returns: |
- Signer instance. |
- |
- Raises: |
- NotImplementedError if they key isn't in PEM format. |
- """ |
- if key.startswith('-----BEGIN '): |
- pkey = RSA.importKey(key) |
- else: |
- raise NotImplementedError( |
- 'PKCS12 format is not supported by the PyCrpto library. ' |
- 'Try converting to a "PEM" ' |
- '(openssl pkcs12 -in xxxxx.p12 -nodes -nocerts > privatekey.pem) ' |
- 'or using PyOpenSSL if native code is an option.') |
- return PyCryptoSigner(pkey) |
- |
-except ImportError: |
- PyCryptoVerifier = None |
- PyCryptoSigner = None |
- |
- |
-if OpenSSLSigner: |
- Signer = OpenSSLSigner |
- Verifier = OpenSSLVerifier |
-elif PyCryptoSigner: |
- Signer = PyCryptoSigner |
- Verifier = PyCryptoVerifier |
-else: |
- raise ImportError('No encryption library found. Please install either ' |
- 'PyOpenSSL, or PyCrypto 2.6 or later') |
- |
- |
-def _urlsafe_b64encode(raw_bytes): |
- return base64.urlsafe_b64encode(raw_bytes).rstrip('=') |
- |
- |
-def _urlsafe_b64decode(b64string): |
- # Guard against unicode strings, which base64 can't handle. |
- b64string = b64string.encode('ascii') |
- padded = b64string + '=' * (4 - len(b64string) % 4) |
- return base64.urlsafe_b64decode(padded) |
- |
- |
-def _json_encode(data): |
- return simplejson.dumps(data, separators = (',', ':')) |
- |
- |
-def make_signed_jwt(signer, payload): |
- """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. |
- |
- Returns: |
- string, The JWT for the payload. |
- """ |
- header = {'typ': 'JWT', 'alg': 'RS256'} |
- |
- segments = [ |
- _urlsafe_b64encode(_json_encode(header)), |
- _urlsafe_b64encode(_json_encode(payload)), |
- ] |
- signing_input = '.'.join(segments) |
- |
- signature = signer.sign(signing_input) |
- segments.append(_urlsafe_b64encode(signature)) |
- |
- logger.debug(str(segments)) |
- |
- return '.'.join(segments) |
- |
- |
-def verify_signed_jwt_with_certs(jwt, certs, audience): |
- """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. |
- """ |
- segments = jwt.split('.') |
- |
- if (len(segments) != 3): |
- raise AppIdentityError( |
- 'Wrong number of segments in token: %s' % jwt) |
- signed = '%s.%s' % (segments[0], segments[1]) |
- |
- signature = _urlsafe_b64decode(segments[2]) |
- |
- # Parse token. |
- json_body = _urlsafe_b64decode(segments[1]) |
- try: |
- parsed = simplejson.loads(json_body) |
- except: |
- raise AppIdentityError('Can\'t parse token: %s' % json_body) |
- |
- # Check signature. |
- verified = False |
- for (keyname, pem) in certs.items(): |
- verifier = Verifier.from_string(pem, True) |
- if (verifier.verify(signed, signature)): |
- verified = True |
- break |
- if not verified: |
- raise AppIdentityError('Invalid token signature: %s' % jwt) |
- |
- # Check creation timestamp. |
- iat = parsed.get('iat') |
- if iat is None: |
- raise AppIdentityError('No iat field in token: %s' % json_body) |
- earliest = iat - CLOCK_SKEW_SECS |
- |
- # Check expiration timestamp. |
- now = long(time.time()) |
- exp = parsed.get('exp') |
- if exp is None: |
- raise AppIdentityError('No exp field in token: %s' % json_body) |
- if exp >= now + MAX_TOKEN_LIFETIME_SECS: |
- raise AppIdentityError( |
- 'exp field too far in future: %s' % json_body) |
- latest = exp + CLOCK_SKEW_SECS |
- |
- if now < earliest: |
- raise AppIdentityError('Token used too early, %d < %d: %s' % |
- (now, earliest, json_body)) |
- if now > latest: |
- raise AppIdentityError('Token used too late, %d > %d: %s' % |
- (now, latest, json_body)) |
- |
- # Check audience. |
- if audience is not None: |
- aud = parsed.get('aud') |
- if aud is None: |
- raise AppIdentityError('No aud field in token: %s' % json_body) |
- if aud != audience: |
- raise AppIdentityError('Wrong recipient, %s != %s: %s' % |
- (aud, audience, json_body)) |
- |
- return parsed |