Python 3 Support

Author: nmurtagh

.travis.yml

@@ -1,6 +1,13 @@
 language: python
 python:
     - "2.7"
+    - "3.3"
+    - "3.4"
+    - "3.5"
+    - "3.5-dev"
+    - "nightly"
+    - "pypy"
 install:
     - pip install tox
-script: tox
+script: 
+    - tox -e travis

jose.py

@@ -1,27 +1,24 @@
+import binascii
+import datetime
 import logging
-logger = logging.getLogger(__name__)
-
-try:
-    from cjson import encode as json_encode, decode as json_decode
-except ImportError:  # pragma: nocover
-    logger.warn('cjson not found, falling back to stdlib json')
-    from json import loads as json_decode, dumps as json_encode
-
+import six
 import zlib
-import datetime
 
 from base64 import urlsafe_b64encode, urlsafe_b64decode
 from collections import namedtuple
 from copy import deepcopy
-from time import time
+from json import loads as json_decode, dumps as json_encode
 from struct import pack
+from time import time
 
 from Crypto.Hash import HMAC, SHA256, SHA384, SHA512
 from Crypto.Cipher import PKCS1_OAEP, AES
 from Crypto.PublicKey import RSA
 from Crypto.Random import get_random_bytes
 from Crypto.Signature import PKCS1_v1_5 as PKCS1_v1_5_SIG
 
+logger = logging.getLogger(__name__)
+
 
 __all__ = ['encrypt', 'decrypt', 'sign', 'verify']
 
@@ -63,7 +60,8 @@
 class Error(Exception):
     """ The base error type raised by jose
     """
-    pass
+    def __init__(self, message):
+        self.message = message
 
 
 class Expired(Error):
@@ -85,7 +83,7 @@ def serialize_compact(jwt):
     :returns: A string, representing the compact serialization of a
               :class:`~jose.JWE` or :class:`~jose.JWS`.
     """
-    return '.'.join(jwt)
+    return six.b('.').join(jwt)
 
 
 def deserialize_compact(jwt):
@@ -95,7 +93,7 @@ def deserialize_compact(jwt):
     :rtype: :class:`~jose.JWT`.
     :raises: :class:`~jose.Error` if the JWT is malformed
     """
-    parts = jwt.split('.')
+    parts = jwt.split(six.b('.'))
 
     # http://tools.ietf.org/html/
     # draft-ietf-jose-json-web-encryption-23#section-9
@@ -109,8 +107,8 @@ def deserialize_compact(jwt):
     return token_type(*parts)
 
 
-def encrypt(claims, jwk, adata='', add_header=None, alg='RSA-OAEP',
-        enc='A128CBC-HS256', rng=get_random_bytes, compression=None):
+def encrypt(claims, jwk, adata=six.b(''), add_header=None, alg='RSA-OAEP',
+            enc='A128CBC-HS256', rng=get_random_bytes, compression=None):
     """ Encrypts the given claims and produces a :class:`~jose.JWE`
 
     :param claims: A `dict` representing the claims for this
@@ -139,14 +137,15 @@ def encrypt(claims, jwk, adata='', add_header=None, alg='RSA-OAEP',
     assert _TEMP_VER_KEY not in claims
     claims[_TEMP_VER_KEY] = _TEMP_VER
 
-    header = dict((add_header or {}).items() + [
-        ('enc', enc), ('alg', alg)])
+    header = dict(
+        list((add_header or {}).items()) + [('enc', enc), ('alg', alg)]
+    )
 
     # promote the temp key to the header
     assert _TEMP_VER_KEY not in header
     header[_TEMP_VER_KEY] = claims[_TEMP_VER_KEY]
 
-    plaintext = json_encode(claims)
+    plaintext = six.b(json_encode(claims))
 
     # compress (if required)
     if compression is not None:
@@ -162,24 +161,29 @@ def encrypt(claims, jwk, adata='', add_header=None, alg='RSA-OAEP',
     ((cipher, _), key_size), ((hash_fn, _), hash_mod) = JWA[enc]
     iv = rng(AES.block_size)
     encryption_key = rng(hash_mod.digest_size)
+    encryption_key_index = hash_mod.digest_size // 2
 
-    ciphertext = cipher(plaintext, encryption_key[-hash_mod.digest_size/2:], iv)
-    hash = hash_fn(_jwe_hash_str(ciphertext, iv, adata),
-            encryption_key[:-hash_mod.digest_size/2], hash_mod)
+    ciphertext = cipher(
+        plaintext, encryption_key[-encryption_key_index:], iv
+    )
+    hash = hash_fn(
+        _jwe_hash_str(ciphertext, iv, adata),
+        encryption_key[:-encryption_key_index], hash_mod
+    )
 
     # cek encryption
     (cipher, _), _ = JWA[alg]
     encryption_key_ciphertext = cipher(encryption_key, jwk)
 
-    return JWE(*map(b64encode_url,
-            (json_encode(header),
-            encryption_key_ciphertext,
-            iv,
-            ciphertext,
-            auth_tag(hash))))
+    jwe_components = (
+        json_encode(header), encryption_key_ciphertext, iv, ciphertext,
+        auth_tag(hash)
+    )
+    return JWE(*map(b64encode_url, jwe_components))
 
 
-def decrypt(jwe, jwk, adata='', validate_claims=True, expiry_seconds=None):
+def decrypt(jwe, jwk, adata=six.b(''), validate_claims=True,
+            expiry_seconds=None):
     """ Decrypts a deserialized :class:`~jose.JWE`
 
     :param jwe: An instance of :class:`~jose.JWE`
@@ -199,8 +203,9 @@ def decrypt(jwe, jwk, adata='', validate_claims=True, expiry_seconds=None):
     :raises: :class:`~jose.Error` if there is an error decrypting the JWE
     """
     header, encryption_key_ciphertext, iv, ciphertext, tag = map(
-        b64decode_url, jwe)
-    header = json_decode(header)
+        b64decode_url, jwe
+    )
+    header = json_decode(header.decode())
 
     # decrypt cek
     (_, decipher), _ = JWA[header['alg']]
@@ -211,9 +216,13 @@ def decrypt(jwe, jwk, adata='', validate_claims=True, expiry_seconds=None):
 
     version = header.get(_TEMP_VER_KEY)
     if version:
-        plaintext = decipher(ciphertext, encryption_key[-mod.digest_size/2:], iv)
-        hash = hash_fn(_jwe_hash_str(ciphertext, iv, adata, version),
-                encryption_key[:-mod.digest_size/2], mod=mod)
+        plaintext = decipher(
+            ciphertext, encryption_key[-mod.digest_size // 2:], iv
+        )
+        hash = hash_fn(
+            _jwe_hash_str(ciphertext, iv, adata, version),
+            encryption_key[:-mod.digest_size // 2], mod=mod
+        )
     else:
         plaintext = decipher(ciphertext, encryption_key[:-mod.digest_size], iv)
         hash = hash_fn(_jwe_hash_str(ciphertext, iv, adata, version),
@@ -231,7 +240,7 @@ def decrypt(jwe, jwk, adata='', validate_claims=True, expiry_seconds=None):
 
         plaintext = decompress(plaintext)
 
-    claims = json_decode(plaintext)
+    claims = json_decode(plaintext.decode())
     try:
         del claims[_TEMP_VER_KEY]
     except KeyError:
@@ -257,11 +266,12 @@ def sign(claims, jwk, add_header=None, alg='HS256'):
     """
     (hash_fn, _), mod = JWA[alg]
 
-    header = dict((add_header or {}).items() + [('alg', alg)])
+    header = dict(list((add_header or {}).items()) + [('alg', alg)])
     header, payload = map(b64encode_url, map(json_encode, (header, claims)))
 
-    sig = b64encode_url(hash_fn(_jws_hash_str(header, payload), jwk['k'],
-        mod=mod))
+    sig = b64encode_url(
+        hash_fn(_jws_hash_str(header, payload), jwk['k'], mod=mod)
+    )
 
     return JWS(header, payload, sig)
 
@@ -285,17 +295,18 @@ def verify(jws, jwk, alg, validate_claims=True, expiry_seconds=None):
     :raises: :class:`~jose.Error` if there is an error decrypting the JWE
     """
     header, payload, sig = map(b64decode_url, jws)
-    header = json_decode(header)
+    header = json_decode(header.decode())
     if alg != header['alg']:
         raise Error('Invalid algorithm')
 
     (_, verify_fn), mod = JWA[header['alg']]
 
-    if not verify_fn(_jws_hash_str(jws.header, jws.payload),
-            jwk['k'], sig, mod=mod):
+    if not verify_fn(
+        _jws_hash_str(jws.header, jws.payload), jwk['k'], sig, mod=mod
+    ):
         raise Error('Mismatched signatures')
 
-    claims = json_decode(b64decode_url(jws.payload))
+    claims = json_decode(b64decode_url(jws.payload).decode())
     _validate(claims, validate_claims, expiry_seconds)
 
     return JWT(header, claims)
@@ -305,27 +316,32 @@ def b64decode_url(istr):
     """ JWT Tokens may be truncated without the usual trailing padding '='
         symbols. Compensate by padding to the nearest 4 bytes.
     """
-    istr = encode_safe(istr)
     try:
-        return urlsafe_b64decode(istr + '=' * (4 - (len(istr) % 4)))
-    except TypeError as e:
+        return urlsafe_b64decode(istr + six.b('=') * (4 - (len(istr) % 4)))
+    except (TypeError, binascii.Error) as e:
         raise Error('Unable to decode base64: %s' % (e))
 
 
 def b64encode_url(istr):
     """ JWT Tokens may be truncated without the usual trailing padding '='
         symbols. Compensate by padding to the nearest 4 bytes.
     """
-    return urlsafe_b64encode(encode_safe(istr)).rstrip('=')
+    return urlsafe_b64encode(encode_safe(istr)).rstrip(six.b('='))
 
 
-def encode_safe(istr, encoding='utf8'):
-    try:
-        return istr.encode(encoding)
-    except UnicodeDecodeError:
-        # this will fail if istr is already encoded
-        pass
-    return istr
+if six.PY3:
+    def encode_safe(istr, encoding='utf8'):
+        if not isinstance(istr, bytes):
+            return bytes(istr, encoding=encoding)
+        return istr
+else:
+    def encode_safe(istr, encoding='utf8'):
+        try:
+            return istr.encode(encoding)
+        except UnicodeDecodeError:
+            # this will fail if istr is already encoded
+            pass
+        return istr
 
 
 def auth_tag(hmac):
@@ -336,11 +352,15 @@ def auth_tag(hmac):
 
 def pad_pkcs7(s):
     sz = AES.block_size - (len(s) % AES.block_size)
-    return s + (chr(sz) * sz)
+    return s + (six.int2byte(sz) * sz)
 
 
-def unpad_pkcs7(s):
-    return s[:-ord(s[-1])]
+if six.PY3:
+    def unpad_pkcs7(s):
+        return s[:-s[-1]]
+else:
+    def unpad_pkcs7(s):
+        return s[:-ord(s[-1])]
 
 
 def encrypt_oaep(plaintext, jwk):
@@ -391,14 +411,24 @@ def decrypt_aescbc(ciphertext, key, iv):
     return unpad_pkcs7(AES.new(key, AES.MODE_CBC, iv).decrypt(ciphertext))
 
 
-def const_compare(stra, strb):
-    if len(stra) != len(strb):
-        return False
+if six.PY3:
+    def const_compare(stra, strb):
+        if len(stra) != len(strb):
+            return False
+
+        res = 0
+        for a, b in zip(stra, strb):
+            res |= a ^ b
+        return res == 0
+else:
+    def const_compare(stra, strb):
+        if len(stra) != len(strb):
+            return False
 
-    res = 0
-    for a, b in zip(stra, strb):
-        res |= ord(a) ^ ord(b)
-    return res == 0
+        res = 0
+        for a, b in zip(stra, strb):
+            res |= ord(a) ^ ord(b)
+        return res == 0
 
 
 class _JWA(object):
@@ -525,34 +555,36 @@ def _validate(claims, validate_claims, expiry_seconds):
         _check_not_before(now, not_before)
 
 
-def _jwe_hash_str(ciphertext, iv, adata='', version=_TEMP_VER):
+def _jwe_hash_str(ciphertext, iv, adata=six.b(''), version=_TEMP_VER):
     # http://tools.ietf.org/html/
     # draft-ietf-jose-json-web-algorithms-24#section-5.2.2.1
     # Both tokens without version and with version 1 should be ignored in
     # the future as they use incorrect hashing. The version parameter
     # should also be removed.
     if not version:
-        return '.'.join((adata, iv, ciphertext, str(len(adata))))
+        return six.b('.').join(
+            (adata, iv, ciphertext, six.b(str(len(adata))))
+        )
     elif version == 1:
-        return '.'.join((adata, iv, ciphertext, pack("!Q", len(adata) * 8)))
-    return ''.join((adata, iv, ciphertext, pack("!Q", len(adata) * 8)))
+        return six.b('.').join(
+            (adata, iv, ciphertext, pack("!Q", len(adata) * 8))
+        )
+    return six.b('').join(
+        (adata, iv, ciphertext, pack("!Q", len(adata) * 8))
+    )
 
 
 def _jws_hash_str(header, claims):
-    return '.'.join((header, claims))
+    return six.b('.').join((header, claims))
 
 
 def cli_decrypt(jwt, key):
-    print decrypt(deserialize_compact(jwt), {'k':key},
-        validate_claims=False)
+    print(decrypt(deserialize_compact(jwt), {'k': key}, validate_claims=False))
 
 
 def _cli():
     import inspect
-    import sys
-
     from argparse import ArgumentParser
-    from copy import copy
 
     parser = ArgumentParser()
     subparsers = parser.add_subparsers(dest='subparser_name')

requirements.txt

@@ -1 +1,2 @@
 pycrypto >= 2.6
+six