Revived PyCrypto and ecdsa-based algorithms as optional jwt.contrib modules.

Author: mark-adams

jwt/contrib/__init__.py

@@ -0,0 +1,58 @@
+# Note: This file is named py_ecdsa.py because import behavior in Python 2
+# would cause ecdsa.py to squash the ecdsa library that it depends upon.
+
+import hashlib
+
+import ecdsa
+
+from jwt.algorithms import Algorithm
+from jwt.compat import string_types, text_type
+
+
+class ECAlgorithm(Algorithm):
+    """
+    Performs signing and verification operations using
+    ECDSA and the specified hash function
+
+    This class requires the ecdsa package to be installed.
+
+    This is based off of the implementation in PyJWT 0.3.2
+    """
+    def __init__(self, hash_alg):
+        self.hash_alg = hash_alg
+
+    SHA256, SHA384, SHA512 = hashlib.sha256, hashlib.sha384, hashlib.sha512
+
+    def prepare_key(self, key):
+
+        if isinstance(key, ecdsa.SigningKey) or \
+           isinstance(key, ecdsa.VerifyingKey):
+            return key
+
+        if isinstance(key, string_types):
+            if isinstance(key, text_type):
+                key = key.encode('utf-8')
+
+            # Attempt to load key. We don't know if it's
+            # a Signing Key or a Verifying Key, so we try
+            # the Verifying Key first.
+            try:
+                key = ecdsa.VerifyingKey.from_pem(key)
+            except ecdsa.der.UnexpectedDER:
+                key = ecdsa.SigningKey.from_pem(key)
+
+        else:
+            raise TypeError('Expecting a PEM-formatted key.')
+
+        return key
+
+    def sign(self, msg, key):
+        return key.sign(msg, hashfunc=self.hash_alg,
+                        sigencode=ecdsa.util.sigencode_der)
+
+    def verify(self, msg, key, sig):
+        try:
+            return key.verify(sig, msg, hashfunc=self.hash_alg,
+                              sigdecode=ecdsa.util.sigdecode_der)
+        except ecdsa.der.UnexpectedDER:
+            return False

jwt/contrib/algorithms/__init__.py

@@ -0,0 +1,46 @@
+import Crypto.Hash.SHA256
+import Crypto.Hash.SHA384
+import Crypto.Hash.SHA512
+
+from Crypto.PublicKey import RSA
+from Crypto.Signature import PKCS1_v1_5
+
+from jwt.algorithms import Algorithm
+from jwt.compat import string_types, text_type
+
+
+class RSAAlgorithm(Algorithm):
+    """
+    Performs signing and verification operations using
+    RSASSA-PKCS-v1_5 and the specified hash function.
+
+    This class requires PyCrypto package to be installed.
+
+    This is based off of the implementation in PyJWT 0.3.2
+    """
+    def __init__(self, hash_alg):
+        self.hash_alg = hash_alg
+
+    SHA256, SHA384, SHA512 = (Crypto.Hash.SHA256, Crypto.Hash.SHA384,
+                              Crypto.Hash.SHA512)
+
+    def prepare_key(self, key):
+
+        if isinstance(key, RSA._RSAobj):
+            return key
+
+        if isinstance(key, string_types):
+            if isinstance(key, text_type):
+                key = key.encode('utf-8')
+
+            key = RSA.importKey(key)
+        else:
+            raise TypeError('Expecting a PEM- or RSA-formatted key.')
+
+        return key
+
+    def sign(self, msg, key):
+        return PKCS1_v1_5.new(key).sign(self.hash_alg.new(msg))
+
+    def verify(self, msg, key, sig):
+        return PKCS1_v1_5.new(key).verify(self.hash_alg.new(msg), sig)

jwt/contrib/algorithms/py_ecdsa.py

@@ -0,0 +1,131 @@
+import base64
+
+from ..compat import unittest
+from ..utils import ensure_bytes, ensure_unicode, key_path
+
+try:
+    from jwt.contrib.algorithms.pycrypto import RSAAlgorithm
+    has_pycrypto = True
+except ImportError:
+    has_pycrypto = False
+
+try:
+    from jwt.contrib.algorithms.py_ecdsa import ECAlgorithm
+    has_ecdsa = True
+except ImportError:
+    has_ecdsa = False
+
+
+@unittest.skipIf(not has_pycrypto, 'Not supported without PyCrypto library')
+class TestPycryptoAlgorithms(unittest.TestCase):
+    def setUp(self):  # noqa
+        pass
+
+    def test_rsa_should_parse_pem_public_key(self):
+        algo = RSAAlgorithm(RSAAlgorithm.SHA256)
+
+        with open(key_path('testkey2_rsa.pub.pem'), 'r') as pem_key:
+            algo.prepare_key(pem_key.read())
+
+    def test_rsa_should_accept_unicode_key(self):
+        algo = RSAAlgorithm(RSAAlgorithm.SHA256)
+
+        with open(key_path('testkey_rsa'), 'r') as rsa_key:
+            algo.prepare_key(ensure_unicode(rsa_key.read()))
+
+    def test_rsa_should_reject_non_string_key(self):
+        algo = RSAAlgorithm(RSAAlgorithm.SHA256)
+
+        with self.assertRaises(TypeError):
+            algo.prepare_key(None)
+
+    def test_rsa_verify_should_return_false_if_signature_invalid(self):
+        algo = RSAAlgorithm(RSAAlgorithm.SHA256)
+
+        jwt_message = ensure_bytes('Hello World!')
+
+        jwt_sig = base64.b64decode(ensure_bytes(
+            'yS6zk9DBkuGTtcBzLUzSpo9gGJxJFOGvUqN01iLhWHrzBQ9ZEz3+Ae38AXp'
+            '10RWwscp42ySC85Z6zoN67yGkLNWnfmCZSEv+xqELGEvBJvciOKsrhiObUl'
+            '2mveSc1oeO/2ujkGDkkkJ2epn0YliacVjZF5+/uDmImUfAAj8lzjnHlzYix'
+            'sn5jGz1H07jYYbi9diixN8IUhXeTafwFg02IcONhum29V40Wu6O5tAKWlJX'
+            'fHJnNUzAEUOXS0WahHVb57D30pcgIji9z923q90p5c7E2cU8V+E1qe8NdCA'
+            'APCDzZZ9zQ/dgcMVaBrGrgimrcLbPjueOKFgSO+SSjIElKA=='))
+
+        jwt_sig += ensure_bytes('123')  # Signature is now invalid
+
+        with open(key_path('testkey_rsa.pub'), 'r') as keyfile:
+            jwt_pub_key = algo.prepare_key(keyfile.read())
+
+        result = algo.verify(jwt_message, jwt_pub_key, jwt_sig)
+        self.assertFalse(result)
+
+    def test_rsa_verify_should_return_true_if_signature_valid(self):
+        algo = RSAAlgorithm(RSAAlgorithm.SHA256)
+
+        jwt_message = ensure_bytes('Hello World!')
+
+        jwt_sig = base64.b64decode(ensure_bytes(
+            'yS6zk9DBkuGTtcBzLUzSpo9gGJxJFOGvUqN01iLhWHrzBQ9ZEz3+Ae38AXp'
+            '10RWwscp42ySC85Z6zoN67yGkLNWnfmCZSEv+xqELGEvBJvciOKsrhiObUl'
+            '2mveSc1oeO/2ujkGDkkkJ2epn0YliacVjZF5+/uDmImUfAAj8lzjnHlzYix'
+            'sn5jGz1H07jYYbi9diixN8IUhXeTafwFg02IcONhum29V40Wu6O5tAKWlJX'
+            'fHJnNUzAEUOXS0WahHVb57D30pcgIji9z923q90p5c7E2cU8V+E1qe8NdCA'
+            'APCDzZZ9zQ/dgcMVaBrGrgimrcLbPjueOKFgSO+SSjIElKA=='))
+
+        with open(key_path('testkey_rsa.pub'), 'r') as keyfile:
+            jwt_pub_key = algo.prepare_key(keyfile.read())
+
+        result = algo.verify(jwt_message, jwt_pub_key, jwt_sig)
+        self.assertTrue(result)
+
+
+@unittest.skipIf(not has_ecdsa, 'Not supported without ecdsa library')
+class TestEcdsaAlgorithms(unittest.TestCase):
+    def test_ec_should_reject_non_string_key(self):
+        algo = ECAlgorithm(ECAlgorithm.SHA256)
+
+        with self.assertRaises(TypeError):
+            algo.prepare_key(None)
+
+    def test_ec_should_accept_unicode_key(self):
+        algo = ECAlgorithm(ECAlgorithm.SHA256)
+
+        with open(key_path('testkey_ec'), 'r') as ec_key:
+            algo.prepare_key(ensure_unicode(ec_key.read()))
+
+    def test_ec_verify_should_return_false_if_signature_invalid(self):
+        algo = ECAlgorithm(ECAlgorithm.SHA256)
+
+        jwt_message = ensure_bytes('Hello World!')
+
+        jwt_sig = base64.b64decode(ensure_bytes(
+            'MIGIAkIB9vYz+inBL8aOTA4auYz/zVuig7TT1bQgKROIQX9YpViHkFa4DT5'
+            '5FuFKn9XzVlk90p6ldEj42DC9YecXHbC2t+cCQgCicY+8f3f/KCNtWK7cif'
+            '6vdsVwm6Lrjs0Ag6ZqCf+olN11hVt1qKBC4lXppqB1gNWEmNQaiz1z2QRyc'
+            'zJ8hSJmbw=='))
+
+        jwt_sig += ensure_bytes('123')  # Signature is now invalid
+
+        with open(key_path('testkey_ec.pub'), 'r') as keyfile:
+            jwt_pub_key = algo.prepare_key(keyfile.read())
+
+        result = algo.verify(jwt_message, jwt_pub_key, jwt_sig)
+        self.assertFalse(result)
+
+    def test_ec_verify_should_return_true_if_signature_valid(self):
+        algo = ECAlgorithm(ECAlgorithm.SHA256)
+
+        jwt_message = ensure_bytes('Hello World!')
+
+        jwt_sig = base64.b64decode(ensure_bytes(
+            'MIGIAkIB9vYz+inBL8aOTA4auYz/zVuig7TT1bQgKROIQX9YpViHkFa4DT5'
+            '5FuFKn9XzVlk90p6ldEj42DC9YecXHbC2t+cCQgCicY+8f3f/KCNtWK7cif'
+            '6vdsVwm6Lrjs0Ag6ZqCf+olN11hVt1qKBC4lXppqB1gNWEmNQaiz1z2QRyc'
+            'zJ8hSJmbw=='))
+
+        with open(key_path('testkey_ec.pub'), 'r') as keyfile:
+            jwt_pub_key = algo.prepare_key(keyfile.read())
+
+        result = algo.verify(jwt_message, jwt_pub_key, jwt_sig)
+        self.assertTrue(result)

jwt/contrib/algorithms/pycrypto.py

@@ -3,7 +3,7 @@
 from jwt.algorithms import Algorithm, HMACAlgorithm
 
 from .compat import unittest
-from .utils import ensure_bytes, ensure_unicode
+from .utils import ensure_bytes, ensure_unicode, key_path
 
 try:
     from jwt.algorithms import RSAAlgorithm, ECAlgorithm
@@ -53,14 +53,14 @@ def test_hmac_should_accept_unicode_key(self):
     def test_rsa_should_parse_pem_public_key(self):
         algo = RSAAlgorithm(RSAAlgorithm.SHA256)
 
-        with open('tests/keys/testkey2_rsa.pub.pem', 'r') as pem_key:
+        with open(key_path('testkey2_rsa.pub.pem'), 'r') as pem_key:
             algo.prepare_key(pem_key.read())
 
     @unittest.skipIf(not has_crypto, 'Not supported without cryptography library')
     def test_rsa_should_accept_unicode_key(self):
         algo = RSAAlgorithm(RSAAlgorithm.SHA256)
 
-        with open('tests/keys/testkey_rsa', 'r') as rsa_key:
+        with open(key_path('testkey_rsa'), 'r') as rsa_key:
             algo.prepare_key(ensure_unicode(rsa_key.read()))
 
     @unittest.skipIf(not has_crypto, 'Not supported without cryptography library')
@@ -84,9 +84,9 @@ def test_rsa_verify_should_return_false_if_signature_invalid(self):
             'fHJnNUzAEUOXS0WahHVb57D30pcgIji9z923q90p5c7E2cU8V+E1qe8NdCA'
             'APCDzZZ9zQ/dgcMVaBrGrgimrcLbPjueOKFgSO+SSjIElKA=='))
 
-        jwt_sig = jwt_sig + ensure_bytes('123')  # Signature is now invalid
+        jwt_sig += ensure_bytes('123')  # Signature is now invalid
 
-        with open('tests/keys/testkey_rsa.pub', 'r') as keyfile:
+        with open(key_path('testkey_rsa.pub'), 'r') as keyfile:
             jwt_pub_key = algo.prepare_key(keyfile.read())
 
         result = algo.verify(jwt_message, jwt_pub_key, jwt_sig)
@@ -106,7 +106,7 @@ def test_rsa_verify_should_return_true_if_signature_valid(self):
             'fHJnNUzAEUOXS0WahHVb57D30pcgIji9z923q90p5c7E2cU8V+E1qe8NdCA'
             'APCDzZZ9zQ/dgcMVaBrGrgimrcLbPjueOKFgSO+SSjIElKA=='))
 
-        with open('tests/keys/testkey_rsa.pub', 'r') as keyfile:
+        with open(key_path('testkey_rsa.pub'), 'r') as keyfile:
             jwt_pub_key = algo.prepare_key(keyfile.read())
 
         result = algo.verify(jwt_message, jwt_pub_key, jwt_sig)
@@ -123,7 +123,7 @@ def test_ec_should_reject_non_string_key(self):
     def test_ec_should_accept_unicode_key(self):
         algo = ECAlgorithm(ECAlgorithm.SHA256)
 
-        with open('tests/keys/testkey_ec', 'r') as ec_key:
+        with open(key_path('testkey_ec'), 'r') as ec_key:
             algo.prepare_key(ensure_unicode(ec_key.read()))
 
     @unittest.skipIf(not has_crypto, 'Not supported without cryptography library')
@@ -138,9 +138,9 @@ def test_ec_verify_should_return_false_if_signature_invalid(self):
             '6vdsVwm6Lrjs0Ag6ZqCf+olN11hVt1qKBC4lXppqB1gNWEmNQaiz1z2QRyc'
             'zJ8hSJmbw=='))
 
-        jwt_sig = ensure_bytes('123')  # Signature is now invalid
+        jwt_sig += ensure_bytes('123')  # Signature is now invalid
 
-        with open('tests/keys/testkey_ec.pub', 'r') as keyfile:
+        with open(key_path('testkey_ec.pub'), 'r') as keyfile:
             jwt_pub_key = algo.prepare_key(keyfile.read())
 
         result = algo.verify(jwt_message, jwt_pub_key, jwt_sig)
@@ -158,7 +158,7 @@ def test_ec_verify_should_return_true_if_signature_valid(self):
             '6vdsVwm6Lrjs0Ag6ZqCf+olN11hVt1qKBC4lXppqB1gNWEmNQaiz1z2QRyc'
             'zJ8hSJmbw=='))
 
-        with open('tests/keys/testkey_ec.pub', 'r') as keyfile:
+        with open(key_path('testkey_ec.pub'), 'r') as keyfile:
             jwt_pub_key = algo.prepare_key(keyfile.read())
 
         result = algo.verify(jwt_message, jwt_pub_key, jwt_sig)

tests/contrib/__init__.py

@@ -1,3 +1,5 @@
+import os
+
 from .compat import text_type
 
 
@@ -13,3 +15,8 @@ def ensure_unicode(key):
         key = key.decode()
 
     return key
+
+
+def key_path(key_name):
+    return os.path.join(os.path.dirname(os.path.realpath(__file__)),
+                        'keys', key_name)

tests/contrib/test_algorithms.py

@@ -1,5 +1,5 @@
 [tox]
-envlist = py26, py27, py32, py33, py34, py34-nocrypto, pep8
+envlist = py26, py27, py27-contrib-crypto, py27-nocrypto, py32, py33, py34, py34-contrib-crypto, py34-nocrypto, pep8
 
 [testenv]
 commands =
@@ -11,6 +11,28 @@ deps =
     unittest2
     coverage
 
+[testenv:py34-contrib-crypto]
+basepython = python3.4
+commands =
+    coverage erase
+    coverage run setup.py test
+    coverage report -m
+deps =
+    pycrypto
+    ecdsa
+    coverage
+
+[testenv:py27-contrib-crypto]
+basepython = python2.7
+commands =
+    coverage erase
+    coverage run setup.py test
+    coverage report -m
+deps =
+    pycrypto
+    ecdsa
+    coverage
+
 [testenv:py34-nocrypto]
 basepython = python3.4
 commands =
@@ -35,6 +57,5 @@ deps =
     flake8
     flake8-import-order
     pep8-naming
-    unittest2
 commands =
     flake8 . --max-line-length=120