Merge pull request #120 from IdentityPython/fernet

Added Fernet encrypter.

Author: rohe

pyproject.toml

@@ -22,7 +22,7 @@ exclude_lines = [
 
 [tool.poetry]
 name = "cryptojwt"
-version = "1.7.1"
+version = "1.8.0"
 description = "Python implementation of JWT, JWE, JWS and JWK"
 authors = ["Roland Hedberg <[email protected]>"]
 license = "Apache-2.0"

src/cryptojwt/jwe/fernet.py

@@ -0,0 +1,51 @@
+import base64
+import os
+from typing import Optional
+from typing import Union
+
+from cryptography.fernet import Fernet
+from cryptography.hazmat.primitives import hashes
+from cryptography.hazmat.primitives.kdf.pbkdf2 import PBKDF2HMAC
+
+from cryptojwt.jwe import Encrypter
+from cryptojwt.utils import as_bytes
+
+DEFAULT_ITERATIONS = 390000
+
+
+class FernetEncrypter(Encrypter):
+    def __init__(
+        self,
+        password: str,
+        salt: Optional[bytes] = "",
+        hash_alg: Optional[str] = "SHA256",
+        digest_size: Optional[int] = 0,
+        iterations: Optional[int] = DEFAULT_ITERATIONS,
+    ):
+        Encrypter.__init__(self)
+        if not salt:
+            salt = os.urandom(16)
+        else:
+            salt = as_bytes(salt)
+
+        _alg = getattr(hashes, hash_alg)
+        # A bit special for SHAKE* and BLAKE* hashes
+        if hash_alg.startswith("SHAKE") or hash_alg.startswith("BLAKE"):
+            _algorithm = _alg(digest_size)
+        else:
+            _algorithm = _alg()
+        kdf = PBKDF2HMAC(algorithm=_algorithm, length=32, salt=salt, iterations=iterations)
+        self.key = base64.urlsafe_b64encode(kdf.derive(as_bytes(password)))
+        self.core = Fernet(self.key)
+
+    def encrypt(self, msg: Union[str, bytes], **kwargs) -> bytes:
+        text = as_bytes(msg)
+        # Padding to block size of AES
+        if len(text) % 16:
+            text += b" " * (16 - len(text) % 16)
+        return self.core.encrypt(as_bytes(text))
+
+    def decrypt(self, msg: Union[str, bytes], **kwargs) -> bytes:
+        dec_text = self.core.decrypt(as_bytes(msg))
+        dec_text = dec_text.rstrip(b" ")
+        return dec_text

src/cryptojwt/jws/jws.py

@@ -116,7 +116,7 @@ def sign_compact(self, keys=None, protected=None, **kwargs):
         key, xargs, _alg = self.alg_keys(keys, "sig", protected)
 
         if "typ" in self:
-            xargs["typ"] = self["typ"]
+            xargs["type"] = self["typ"]
 
         _headers.update(xargs)
         jwt = JWSig(**_headers)

tests/test_07_jwe.py

@@ -21,6 +21,7 @@
 from cryptojwt.jwe.exception import NoSuitableEncryptionKey
 from cryptojwt.jwe.exception import UnsupportedBitLength
 from cryptojwt.jwe.exception import WrongEncryptionAlgorithm
+from cryptojwt.jwe.fernet import FernetEncrypter
 from cryptojwt.jwe.jwe import JWE
 from cryptojwt.jwe.jwe import factory
 from cryptojwt.jwe.jwe_ec import JWE_EC
@@ -138,7 +139,8 @@ def test_jwe_09_a1():
 
     b64_ejek = (
         b"ApfOLCaDbqs_JXPYy2I937v_xmrzj"
-        b"-Iss1mG6NAHmeJViM6j2l0MHvfseIdHVyU2BIoGVu9ohvkkWiRq5DL2jYZTPA9TAdwq3FUIVyoH-Pedf6elHIVFi2KGDEspYMtQARMMSBcS7pslx6flh1Cfh3GBKysztVMEhZ_maFkm4PYVCsJsvq6Ct3fg2CJPOs0X1DHuxZKoIGIqcbeK4XEO5a0h5TAuJObKdfO0dKwfNSSbpu5sFrpRFwV2FTTYoqF4zI46N9-_hMIznlEpftRXhScEJuZ9HG8C8CHB1WRZ_J48PleqdhF4o7fB5J1wFqUXBtbtuGJ_A2Xe6AEhrlzCOw"
+        b"-Iss1mG6NAHmeJViM6j2l0MHvfseIdHVyU2BIoGVu9ohvkkWiRq5DL2jYZTPA9TAdwq3FUIVyoH"
+        b"-Pedf6elHIVFi2KGDEspYMtQARMMSBcS7pslx6flh1Cfh3GBKysztVMEhZ_maFkm4PYVCsJsvq6Ct3fg2CJPOs0X1DHuxZKoIGIqcbeK4XEO5a0h5TAuJObKdfO0dKwfNSSbpu5sFrpRFwV2FTTYoqF4zI46N9-_hMIznlEpftRXhScEJuZ9HG8C8CHB1WRZ_J48PleqdhF4o7fB5J1wFqUXBtbtuGJ_A2Xe6AEhrlzCOw"
     )
 
     iv = intarr2bytes([227, 197, 117, 252, 2, 219, 233, 68, 180, 225, 77, 219])
@@ -243,7 +245,8 @@ def test_jwe_09_a1():
         [
             b"eyJhbGciOiJSU0EtT0FFUCIsImVuYyI6IkEyNTZHQ00ifQ",
             b"ApfOLCaDbqs_JXPYy2I937v_xmrzj"
-            b"-Iss1mG6NAHmeJViM6j2l0MHvfseIdHVyU2BIoGVu9ohvkkWiRq5DL2jYZTPA9TAdwq3FUIVyoH-Pedf6elHIVFi2KGDEspYMtQARMMSBcS7pslx6flh1Cfh3GBKysztVMEhZ_maFkm4PYVCsJsvq6Ct3fg2CJPOs0X1DHuxZKoIGIqcbeK4XEO5a0h5TAuJObKdfO0dKwfNSSbpu5sFrpRFwV2FTTYoqF4zI46N9-_hMIznlEpftRXhScEJuZ9HG8C8CHB1WRZ_J48PleqdhF4o7fB5J1wFqUXBtbtuGJ_A2Xe6AEhrlzCOw",
+            b"-Iss1mG6NAHmeJViM6j2l0MHvfseIdHVyU2BIoGVu9ohvkkWiRq5DL2jYZTPA9TAdwq3FUIVyoH"
+            b"-Pedf6elHIVFi2KGDEspYMtQARMMSBcS7pslx6flh1Cfh3GBKysztVMEhZ_maFkm4PYVCsJsvq6Ct3fg2CJPOs0X1DHuxZKoIGIqcbeK4XEO5a0h5TAuJObKdfO0dKwfNSSbpu5sFrpRFwV2FTTYoqF4zI46N9-_hMIznlEpftRXhScEJuZ9HG8C8CHB1WRZ_J48PleqdhF4o7fB5J1wFqUXBtbtuGJ_A2Xe6AEhrlzCOw",
             b"48V1_ALb6US04U3b",
             b"5eym8TW_c8SuK0ltJ3rpYIzOeDQz7TALvtu6UG9oMo4vpzs9tX_EFShS8iB7j6jiSdiwkIr3ajwQzaBtQD_A",
             b"ghEgxninkHEAMp4xZtB2mA",
@@ -643,3 +646,38 @@ def test_invalid():
     decrypter = JWE(plain, alg="A128KW", enc="A128CBC-HS256")
     with pytest.raises(BadSyntax):
         decrypter.decrypt("a.b.c.d.e", keys=[encryption_key])
+
+
+def test_fernet():
+    encryption_key = SYMKey(use="enc", key="DukeofHazardpass", kid="some-key-id")
+
+    encrypter = FernetEncrypter(encryption_key.key)
+    _token = encrypter.encrypt(plain)
+
+    decrypter = encrypter
+    resp = decrypter.decrypt(_token)
+    assert resp == plain
+
+
+def test_fernet_sha512():
+    encryption_key = SYMKey(use="enc", key="DukeofHazardpass", kid="some-key-id")
+
+    encrypter = FernetEncrypter(encryption_key.key, hash_alg="SHA512")
+    _token = encrypter.encrypt(plain)
+
+    decrypter = encrypter
+    resp = decrypter.decrypt(_token)
+    assert resp == plain
+
+
+def test_fernet_blake2s():
+    encryption_key = SYMKey(use="enc", key="DukeofHazardpass", kid="some-key-id")
+
+    encrypter = FernetEncrypter(
+        encryption_key.key, hash_alg="BLAKE2s", digest_size=32, iterations=1000
+    )
+    _token = encrypter.encrypt(plain)
+
+    decrypter = encrypter
+    resp = decrypter.decrypt(_token)
+    assert resp == plain