Implement support with PyCryptodome

Author: jpadilla

jwt/contrib/algorithms/pycryptodome.py

@@ -0,0 +1,90 @@
+import Cryptodome.Hash.SHA256
+import Cryptodome.Hash.SHA384
+import Cryptodome.Hash.SHA512
+from Cryptodome.PublicKey import ECC, RSA
+from Cryptodome.Signature import DSS, 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 PyCryptodome package to be installed.
+    """
+
+    SHA256 = Cryptodome.Hash.SHA256
+    SHA384 = Cryptodome.Hash.SHA384
+    SHA512 = Cryptodome.Hash.SHA512
+
+    def __init__(self, hash_alg):
+        self.hash_alg = hash_alg
+
+    def prepare_key(self, key):
+
+        if isinstance(key, RSA.RsaKey):
+            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)
+
+
+class ECAlgorithm(Algorithm):
+    """
+    Performs signing and verification operations using
+    ECDSA and the specified hash function
+
+    This class requires the PyCryptodome package to be installed.
+    """
+
+    SHA256 = Cryptodome.Hash.SHA256
+    SHA384 = Cryptodome.Hash.SHA384
+    SHA512 = Cryptodome.Hash.SHA512
+
+    def __init__(self, hash_alg):
+        self.hash_alg = hash_alg
+
+    def prepare_key(self, key):
+
+        if isinstance(key, ECC.EccKey):
+            return key
+
+        if isinstance(key, string_types):
+            if isinstance(key, text_type):
+                key = key.encode("utf-8")
+            key = ECC.import_key(key)
+        else:
+            raise TypeError("Expecting a PEM- or ECC-formatted key.")
+
+        return key
+
+    def sign(self, msg, key):
+        signer = DSS.new(key, "fips-186-3")
+        hash_obj = self.hash_alg.new(msg)
+        return signer.sign(hash_obj)
+
+    def verify(self, msg, key, sig):
+        verifier = DSS.new(key, "fips-186-3")
+        hash_obj = self.hash_alg.new(msg)
+
+        try:
+            verifier.verify(hash_obj, sig)
+            return True
+        except ValueError:
+            return False

pyproject.toml

@@ -11,4 +11,4 @@ use_parentheses=true
 combine_as_imports=true
 
 known_first_party="jwt"
-known_third_party=["Crypto", "ecdsa", "pytest", "setuptools", "sphinx_rtd_theme"]
+known_third_party=["Crypto", "Cryptodome", "ecdsa", "pytest", "setuptools", "sphinx_rtd_theme"]

setup.py

@@ -35,13 +35,15 @@ def get_version(package):
 EXTRAS_REQUIRE = {
     "jwks-client": ["requests"],
     "tests": ["pytest>=4.0.1,<5.0.0", "pytest-cov>=2.6.0,<3.0.0"],
-    "crypto": ["cryptography >= 1.4"],
+    "cryptography": ["cryptography >= 1.4"],
+    "pycryptodome": ["pycryptodomex"],
 }
 
 EXTRAS_REQUIRE["dev"] = (
     EXTRAS_REQUIRE["tests"]
-    + EXTRAS_REQUIRE["crypto"]
+    + EXTRAS_REQUIRE["cryptography"]
     + EXTRAS_REQUIRE["jwks-client"]
+    + EXTRAS_REQUIRE["pycryptodome"]
     + ["mypy", "pre-commit"]
 )
 

tests/contrib/test_algorithms.py

@@ -20,6 +20,15 @@
 except ImportError:
     has_ecdsa = False
 
+try:
+    # fmt: off
+    from jwt.contrib.algorithms.pycryptodome import RSAAlgorithm, ECAlgorithm  # noqa: F811
+    # fmt: on
+
+    has_pycryptodome = True
+except ImportError:
+    has_pycryptodome = False
+
 
 @pytest.mark.skipif(
     not has_pycrypto, reason="Not supported without PyCrypto library"
@@ -212,3 +221,188 @@ def test_ec_prepare_key_should_be_idempotent(self):
             jwt_pub_key_second = algo.prepare_key(jwt_pub_key_first)
 
         assert jwt_pub_key_first == jwt_pub_key_second
+
+
+@pytest.mark.skipif(
+    not has_pycryptodome, reason="Not supported without PyCryptodome library"
+)
+class TestPyCryptodomeAlgorithms:
+    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(force_unicode(rsa_key.read()))
+
+    def test_rsa_should_reject_non_string_key(self):
+        algo = RSAAlgorithm(RSAAlgorithm.SHA256)
+
+        with pytest.raises(TypeError):
+            algo.prepare_key(None)
+
+    def test_rsa_sign_should_generate_correct_signature_value(self):
+        algo = RSAAlgorithm(RSAAlgorithm.SHA256)
+
+        jwt_message = force_bytes("Hello World!")
+
+        expected_sig = base64.b64decode(
+            force_bytes(
+                "yS6zk9DBkuGTtcBzLUzSpo9gGJxJFOGvUqN01iLhWHrzBQ9ZEz3+Ae38AXp"
+                "10RWwscp42ySC85Z6zoN67yGkLNWnfmCZSEv+xqELGEvBJvciOKsrhiObUl"
+                "2mveSc1oeO/2ujkGDkkkJ2epn0YliacVjZF5+/uDmImUfAAj8lzjnHlzYix"
+                "sn5jGz1H07jYYbi9diixN8IUhXeTafwFg02IcONhum29V40Wu6O5tAKWlJX"
+                "fHJnNUzAEUOXS0WahHVb57D30pcgIji9z923q90p5c7E2cU8V+E1qe8NdCA"
+                "APCDzZZ9zQ/dgcMVaBrGrgimrcLbPjueOKFgSO+SSjIElKA=="
+            )
+        )
+
+        with open(key_path("testkey_rsa"), "r") as keyfile:
+            jwt_key = algo.prepare_key(keyfile.read())
+
+        with open(key_path("testkey_rsa.pub"), "r") as keyfile:
+            jwt_pub_key = algo.prepare_key(keyfile.read())
+
+        algo.sign(jwt_message, jwt_key)
+        result = algo.verify(jwt_message, jwt_pub_key, expected_sig)
+        assert result
+
+    def test_rsa_verify_should_return_false_if_signature_invalid(self):
+        algo = RSAAlgorithm(RSAAlgorithm.SHA256)
+
+        jwt_message = force_bytes("Hello World!")
+
+        jwt_sig = base64.b64decode(
+            force_bytes(
+                "yS6zk9DBkuGTtcBzLUzSpo9gGJxJFOGvUqN01iLhWHrzBQ9ZEz3+Ae38AXp"
+                "10RWwscp42ySC85Z6zoN67yGkLNWnfmCZSEv+xqELGEvBJvciOKsrhiObUl"
+                "2mveSc1oeO/2ujkGDkkkJ2epn0YliacVjZF5+/uDmImUfAAj8lzjnHlzYix"
+                "sn5jGz1H07jYYbi9diixN8IUhXeTafwFg02IcONhum29V40Wu6O5tAKWlJX"
+                "fHJnNUzAEUOXS0WahHVb57D30pcgIji9z923q90p5c7E2cU8V+E1qe8NdCA"
+                "APCDzZZ9zQ/dgcMVaBrGrgimrcLbPjueOKFgSO+SSjIElKA=="
+            )
+        )
+
+        jwt_sig += force_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)
+        assert not result
+
+    def test_rsa_verify_should_return_true_if_signature_valid(self):
+        algo = RSAAlgorithm(RSAAlgorithm.SHA256)
+
+        jwt_message = force_bytes("Hello World!")
+
+        jwt_sig = base64.b64decode(
+            force_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)
+        assert result
+
+    def test_rsa_prepare_key_should_be_idempotent(self):
+        algo = RSAAlgorithm(RSAAlgorithm.SHA256)
+
+        with open(key_path("testkey_rsa.pub"), "r") as keyfile:
+            jwt_pub_key_first = algo.prepare_key(keyfile.read())
+            jwt_pub_key_second = algo.prepare_key(jwt_pub_key_first)
+
+        assert jwt_pub_key_first == jwt_pub_key_second
+
+    def test_ec_should_reject_non_string_key(self):
+        algo = ECAlgorithm(ECAlgorithm.SHA256)
+
+        with pytest.raises(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(force_unicode(ec_key.read()))
+
+    def test_ec_sign_should_generate_correct_signature_value(self):
+        algo = ECAlgorithm(ECAlgorithm.SHA256)
+
+        jwt_message = force_bytes("Hello World!")
+
+        expected_sig = base64.b64decode(
+            force_bytes(
+                "v163MdgPbEj2C/ZPRvPtLBtKaKitqZ4wg0Vs7mpoQb+i18flD"
+                "eKb19cZn7h1E3tuP1CgthJvvIBdRKQf0OXlZA=="
+            )
+        )
+
+        with open(key_path("testkey_ec"), "r") as keyfile:
+            jwt_key = algo.prepare_key(keyfile.read())
+
+        with open(key_path("testkey_ec.pub"), "r") as keyfile:
+            jwt_pub_key = algo.prepare_key(keyfile.read())
+
+        algo.sign(jwt_message, jwt_key)
+        result = algo.verify(jwt_message, jwt_pub_key, expected_sig)
+        assert result
+
+    def test_ec_verify_should_return_false_if_signature_invalid(self):
+        algo = ECAlgorithm(ECAlgorithm.SHA256)
+
+        jwt_message = force_bytes("Hello World!")
+
+        jwt_sig = base64.b64decode(
+            force_bytes(
+                "v163MdgPbEj2C/ZPRvPtLBtKaKitqZ4wg0Vs7mpoQb+i18flD"
+                "eKb19cZn7h1E3tuP1CgthJvvIBdRKQf0OXlZA=="
+            )
+        )
+
+        jwt_sig += force_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)
+        assert not result
+
+    def test_ec_verify_should_return_true_if_signature_valid(self):
+        algo = ECAlgorithm(ECAlgorithm.SHA256)
+
+        jwt_message = force_bytes("Hello World!")
+
+        jwt_sig = base64.b64decode(
+            force_bytes(
+                "v163MdgPbEj2C/ZPRvPtLBtKaKitqZ4wg0Vs7mpoQb+i18flD"
+                "eKb19cZn7h1E3tuP1CgthJvvIBdRKQf0OXlZA=="
+            )
+        )
+
+        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)
+        assert result
+
+    def test_ec_prepare_key_should_be_idempotent(self):
+        algo = ECAlgorithm(ECAlgorithm.SHA256)
+
+        with open(key_path("testkey_ec.pub"), "r") as keyfile:
+            jwt_pub_key_first = algo.prepare_key(keyfile.read())
+            jwt_pub_key_second = algo.prepare_key(jwt_pub_key_first)
+
+        assert jwt_pub_key_first == jwt_pub_key_second

tests/keys/testkey_ec

@@ -1,7 +1,5 @@
------BEGIN EC PRIVATE KEY-----
-MIHbAgEBBEG4xN/z6gk7bPkEzs1hHOsbs+Gi2lku8YH4LkS4E1q9U9jSOjvEcFNH
-m/CQjKi1rtpAb0/WL3p/wXsc26e7zmAA5KAHBgUrgQQAI6GBiQOBhgAEAVnCcDxA
-J0v5OJBYFIcTReydEkEIWRvpzYMvv5l8IUOT2SFJiHdWtU45DV4is7+g6bbQanbh
-28/1dBLR/kH1stAeAYWeTJ08gxo3M9Q0KinXsXm4c6G24UiGY6WHeWlOPKPa16fz
-pwJ62o3XaRrCdGzX+K7TCwahWCTeizrJQAe8UwUY
------END EC PRIVATE KEY-----
+-----BEGIN PRIVATE KEY-----
+MIGHAgEAMBMGByqGSM49AgEGCCqGSM49AwEHBG0wawIBAQQg2nninfu2jMHDwAbn
+9oERUhRADS6duQaJEadybLaa0YShRANCAAQfMBxRZKUYEdy5/fLdGI2tYj6kTr50
+PZPt8jOD23rAR7dhtNpG1ojqopmH0AH5wEXadgk8nLCT4cAPK59Qp9Ek
+-----END PRIVATE KEY-----

tests/keys/testkey_ec.pub

@@ -1,6 +1,4 @@
 -----BEGIN PUBLIC KEY-----
-MIGbMBAGByqGSM49AgEGBSuBBAAjA4GGAAQBWcJwPEAnS/k4kFgUhxNF7J0SQQhZ
-G+nNgy+/mXwhQ5PZIUmId1a1TjkNXiKzv6DpttBqduHbz/V0EtH+QfWy0B4BhZ5M
-nTyDGjcz1DQqKdexebhzobbhSIZjpYd5aU48o9rXp/OnAnrajddpGsJ0bNf4rtML
-BqFYJN6LOslAB7xTBRg=
+MFkwEwYHKoZIzj0CAQYIKoZIzj0DAQcDQgAEHzAcUWSlGBHcuf3y3RiNrWI+pE6+
+dD2T7fIzg9t6wEe3YbTaRtaI6qKZh9AB+cBF2nYJPJywk+HADyufUKfRJA==
 -----END PUBLIC KEY-----

tests/keys/testkey_ec_ssh.pub

@@ -1 +1 @@
-ecdsa-sha2-nistp521 AAAAE2VjZHNhLXNoYTItbmlzdHA1MjEAAAAIbmlzdHA1MjEAAACFBAFZwnA8QCdL+TiQWBSHE0XsnRJBCFkb6c2DL7+ZfCFDk9khSYh3VrVOOQ1eIrO/oOm20Gp24dvP9XQS0f5B9bLQHgGFnkydPIMaNzPUNCop17F5uHOhtuFIhmOlh3lpTjyj2ten86cCetqN12kawnRs1/iu0wsGoVgk3os6yUAHvFMFGA==
+ecdsa-sha2-nistp256 AAAAE2VjZHNhLXNoYTItbmlzdHAyNTYAAAAIbmlzdHAyNTYAAABBBB8wHFFkpRgR3Ln98t0Yja1iPqROvnQ9k+3yM4PbesBHt2G02kbWiOqimYfQAfnARdp2CTycsJPhwA8rn1Cn0SQ=

tests/test_api_jws.py

@@ -224,12 +224,9 @@ def test_decodes_valid_es384_jws(self, jws):
         with open("tests/keys/testkey_ec.pub", "r") as fp:
             example_pubkey = fp.read()
         example_jws = (
-            b"eyJhbGciOiJFUzM4NCIsInR5cCI6IkpXVCJ9"
-            b".eyJoZWxsbyI6IndvcmxkIn0"
-            b".AGtlemKghaIaYh1yeeekFH9fRuNY7hCaw5hUgZ5aG1N"
-            b"2F8FIbiKLaZKr8SiFdTimXFVTEmxpBQ9sRmdsDsnrM-1"
-            b"HAG0_zxxu0JyINOFT2iqF3URYl9HZ8kZWMeZAtXmn6Cw"
-            b"PXRJD2f7N-f7bJ5JeL9VT5beI2XD3FlK3GgRvI-eE-2Ik"
+            b"eyJhbGciOiJFUzI1NiIsInR5cCI6IkpXVCJ9."
+            b"eyJoZWxsbyI6IndvcmxkIn0.TORyNQab_MoXM7DvNKaTwbrJr4UY"
+            b"d2SsX8hhlnWelQFmPFSf_JzC2EbLnar92t-bXsDovzxp25ExazrVHkfPkQ"
         )
         decoded_payload = jws.decode(example_jws, example_pubkey)
         json_payload = json.loads(force_unicode(decoded_payload))

tests/test_api_jwt.py

@@ -199,20 +199,17 @@ def test_encode_datetime(self, jwt):
     @pytest.mark.skipif(
         not has_crypto, reason="Can't run without cryptography library"
     )
-    def test_decodes_valid_es384_jwt(self, jwt):
+    def test_decodes_valid_es256_jwt(self, jwt):
         example_payload = {"hello": "world"}
         with open("tests/keys/testkey_ec.pub", "r") as fp:
             example_pubkey = fp.read()
         example_jwt = (
-            b"eyJhbGciOiJFUzM4NCIsInR5cCI6IkpXVCJ9"
-            b".eyJoZWxsbyI6IndvcmxkIn0"
-            b".AddMgkmRhzqptDYqlmy_f2dzM6O9YZmVo-txs_CeAJD"
-            b"NoD8LN7YiPeLmtIhkO5_VZeHHKvtQcGc4lsq-Y72c4dK"
-            b"pANr1f6HEYhjpBc03u_bv06PYMcr5N2-9k97-qf-JCSb"
-            b"zqW6R250Q7gNCX5R7NrCl7MTM4DTBZkGbUlqsFUleiGlj"
+            b"eyJhbGciOiJFUzI1NiIsInR5cCI6IkpXVCJ9."
+            b"eyJoZWxsbyI6IndvcmxkIn0.TORyNQab_MoXM7DvNKaTwbrJr4UY"
+            b"d2SsX8hhlnWelQFmPFSf_JzC2EbLnar92t-bXsDovzxp25ExazrVHkfPkQ"
         )
-        decoded_payload = jwt.decode(example_jwt, example_pubkey)
 
+        decoded_payload = jwt.decode(example_jwt, example_pubkey)
         assert decoded_payload == example_payload
 
     # 'Control' RSA JWT created by another library.