Cleaned up so all _decrypt respond in the same manner.

Author: rohe

src/jwkest/__init__.py

@@ -14,7 +14,7 @@
 
 from binascii import unhexlify
 
-__version__ = "1.3.5"
+__version__ = "1.4.0"
 
 logger = logging.getLogger(__name__)
 

src/jwkest/jwe.py

@@ -387,9 +387,9 @@ def _decrypt(enc, key, ctxt, auth_data, iv, tag):
             try:
                 text = gcm.decrypt(bytes_to_long(iv), ctxt, bytes_to_long(tag),
                                    auth_data)
-                return text, True
+                return text
             except DecryptionFailed:
-                return None, False
+                raise
         elif enc in ["A128CBC-HS256", "A192CBC-HS384", "A256CBC-HS512"]:
             return aes_cbc_hmac_decrypt(key, iv, auth_data, ctxt, tag)
         else:
@@ -563,12 +563,10 @@ def decrypt(self, token, key, cek=None):
         except AssertionError:
             raise NotSupportedAlgorithm(enc)
 
-        msg, flag = self._decrypt(enc, cek, jwe.ciphertext(),
-                                  jwe.b64_protected_header(),
-                                  jwe.initialization_vector(),
-                                  jwe.authentication_tag())
-        if flag is False:
-            raise DecryptionFailed()
+        msg = self._decrypt(enc, cek, jwe.ciphertext(),
+                            jwe.b64_protected_header(),
+                            jwe.initialization_vector(),
+                            jwe.authentication_tag())
 
         if "zip" in jwe.headers and jwe.headers["zip"] == "DEF":
             msg = zlib.decompress(msg)
@@ -603,7 +601,8 @@ def enc_setup(self, msg, auth_data, key=None, **kwargs):
         # Generate an ephemeral key pair if none is given
         curve = NISTEllipticCurve.by_name(key.crv)
         if "epk" in kwargs:
-            epk = kwargs["epk"] if isinstance(kwargs["epk"], ECKey) else ECKey(kwargs["epk"])
+            epk = kwargs["epk"] if isinstance(kwargs["epk"], ECKey) else ECKey(
+                kwargs["epk"])
         else:
             epk = ECKey().load_key(key=NISTEllipticCurve.by_name(key.crv))
 
@@ -650,7 +649,8 @@ def dec_setup(self, token, key=None, **kwargs):
 
         # Handle EPK / Curve
         if "epk" not in self.headers or "crv" not in self.headers["epk"]:
-            raise Exception("Ephemeral Public Key Missing in ECDH-ES Computation")
+            raise Exception(
+                "Ephemeral Public Key Missing in ECDH-ES Computation")
 
         epubkey = ECKey(**self.headers["epk"])
         apu = apv = ""
@@ -716,12 +716,12 @@ def decrypt(self, token=None, key=None, **kwargs):
         if not self.cek:
             raise Exception("Content Encryption Key is Not Yet Set")
 
-        msg, valid = super(JWE_EC, self)._decrypt(self.headers["enc"], self.cek,
-                                                  self.ctxt,
-                                                  jwe.b64part[0],
-                                                  self.iv, self.tag)
+        msg = super(JWE_EC, self)._decrypt(self.headers["enc"], self.cek,
+                                           self.ctxt,
+                                           jwe.b64part[0],
+                                           self.iv, self.tag)
         self.msg = msg
-        self.msg_valid = valid
+        self.msg_valid = True
         return msg
 
 
@@ -782,7 +782,9 @@ def encrypt(self, keys=None, cek="", iv="", **kwargs):
 
         if not keys:
             logger.error(
-                "Could not find any suitable encryption key for alg='{}'".format(_alg))
+                "Could not find any suitable encryption key for alg='{"
+                "}'".format(
+                    _alg))
             raise NoSuitableEncryptionKey(_alg)
 
         # Determine Encryption Class by Algorithm

tests/test_4_jwe.py

@@ -193,8 +193,8 @@ def full_path(local_file):
 rsa = RSA.importKey(open(KEY, 'r').read())
 plain = b'Now is the time for all good men to come to the aid of their country.'
 
-def test_cek_reuse_encryption_rsaes_rsa15():
 
+def test_cek_reuse_encryption_rsaes_rsa15():
     _rsa = JWE_RSA(plain, alg="RSA1_5", enc="A128CBC-HS256")
     jwt = _rsa.encrypt(rsa)
     dec = JWE_RSA()
@@ -209,8 +209,8 @@ def test_cek_reuse_encryption_rsaes_rsa15():
 
     assert msg == plain
 
-def test_cek_reuse_encryption_rsaes_rsa_oaep():
 
+def test_cek_reuse_encryption_rsaes_rsa_oaep():
     _rsa = JWE_RSA(plain, alg="RSA-OAEP", enc="A256GCM")
     jwt = _rsa.encrypt(rsa)
     dec = JWE_RSA()
@@ -225,6 +225,7 @@ def test_cek_reuse_encryption_rsaes_rsa_oaep():
 
     assert msg == plain
 
+
 def test_rsa_encrypt_decrypt_rsa_cbc():
     _rsa = JWE_RSA(plain, alg="RSA1_5", enc="A128CBC-HS256")
     jwt = _rsa.encrypt(rsa)
@@ -239,8 +240,8 @@ def test_rsa_encrypt_decrypt_rsa_oaep_gcm():
     msg = JWE_RSA().decrypt(jwt, rsa)
 
     assert msg == plain
-    
-    
+
+
 def test_rsa_encrypt_decrypt_rsa_oaep_256_gcm():
     jwt = JWE_RSA(plain[:1], alg="RSA-OAEP-256", enc="A256GCM").encrypt(rsa)
     msg = JWE_RSA().decrypt(jwt, rsa)
@@ -277,10 +278,11 @@ def test_rsa_with_kid():
 localpriv, localpub = curve.key_pair()
 
 localkey = ECKey(crv=curve.name(), d=localpriv, x=localpub[0], y=localpub[1])
-remotekey = ECKey(crv=curve.name(), d=remotepriv, x=remotepub[0], y=remotepub[1])
+remotekey = ECKey(crv=curve.name(), d=remotepriv, x=remotepub[0],
+                  y=remotepub[1])
 
-def test_ecdh_encrypt_decrypt_direct_key():
 
+def test_ecdh_encrypt_decrypt_direct_key():
     jwenc = JWE_EC(plain, alg="ECDH-ES", enc="A128GCM")
     cek, encrypted_key, iv, params, ret_epk = jwenc.enc_setup(plain, '',
                                                               key=remotekey,
@@ -333,13 +335,14 @@ def test_ecdh_encrypt_decrypt_keywrapped_key():
 
 
 def test_sym_encrypt_decrypt():
-    encryption_key = SYMKey(use="enc", key='DukeofHazardpass', kid="some-key-id")
+    encryption_key = SYMKey(use="enc", key='DukeofHazardpass',
+                            kid="some-key-id")
     jwe = JWE_SYM("some content", alg="A128KW", enc="A128CBC-HS256")
     _jwe = jwe.encrypt(key=encryption_key, kid="some-key-id")
     jwdec = JWE_SYM()
 
     resp = jwdec.decrypt(_jwe, encryption_key)
-    assert resp[0] == b'some content'
+    assert resp == b'some content'
 
 
 def test_ecdh_no_setup_dynamic_epk():
@@ -348,4 +351,4 @@ def test_ecdh_no_setup_dynamic_epk():
     assert jwt
     ret_jwe = factory(jwt)
     res = ret_jwe.decrypt(jwt, [remotekey])
-    assert res
+    assert res