Merge branch 'master' of github.com:rohe/pyjwkest

Author: Roland Hedberg

src/jwkest/__init__.py

@@ -35,6 +35,10 @@ class Invalid(JWKESTException):
     """The JWT is invalid."""
 
 
+class WrongNumberOfParts(Invalid):
+    pass
+
+
 class BadSyntax(Invalid):
     """The JWT could not be parsed because the syntax is invalid."""
 

src/jwkest/jwe.py

@@ -18,7 +18,7 @@
 from Crypto.Cipher import PKCS1_v1_5
 from Crypto.Cipher import PKCS1_OAEP
 
-from jwkest import b64d, as_bytes
+from jwkest import b64d, as_bytes, WrongNumberOfParts
 from jwkest import b64e
 from jwkest import JWKESTException
 from jwkest import MissingKey
@@ -307,6 +307,9 @@ def is_jwe(self):
                     return False
         return True
 
+    def __len__(self):
+        return len(self.part)
+
 
 class JWe(JWx):
     @staticmethod
@@ -430,6 +433,9 @@ def decrypt(self, token, key=None, cek=None):
 
         jwe = JWEnc().unpack(token)
 
+        if len(jwe) != 5:
+            raise WrongNumberOfParts(len(jwe))
+
         if not cek:
             jek = jwe.encrypted_key()
             # The iv for this function must be 64 bit

src/jwkest/jws.py

@@ -23,7 +23,7 @@
 from Crypto.Util.number import bytes_to_long
 import sys
 
-from jwkest import b64d, as_unicode
+from jwkest import b64d, as_unicode, Invalid, WrongNumberOfParts
 from jwkest import b64e
 from jwkest import constant_time_compare
 from jwkest import safe_str_cmp
@@ -219,6 +219,15 @@ def sign_input(self):
     def signature(self):
         return self.part[2]
 
+    def __len__(self):
+        return len(self.part)
+
+    def valid(self):
+        if len(self) != 3:
+            return False
+
+        return True
+
 
 class JWx(object):
     args = ["alg", "jku", "jwk", "x5u", "x5t", "x5c", "kid", "typ", "cty",
@@ -483,6 +492,9 @@ def verify_compact(self, jws, keys=None, allow_none=False, sigalg=None):
         :return:
         """
         jwt = JWSig().unpack(jws)
+        if len(jwt) != 3:
+            raise WrongNumberOfParts(len(jwt))
+
         self.jwt = jwt
 
         try:
@@ -528,7 +540,7 @@ def verify_compact(self, jws, keys=None, allow_none=False, sigalg=None):
             try:
                 res = verifier.verify(jwt.sign_input(), jwt.signature(),
                                       key.get_key(alg=_alg, private=False))
-            except BadSignature:
+            except (BadSignature, IndexError):
                 pass
             else:
                 if res is True:

tests/test_3_jws.py

@@ -487,5 +487,24 @@ def test_dj_usage():
     _jwt = factory(sjwt)
     assert _jwt.jwt.headers['alg'] == 'RS256'
 
+def test_rs256_rm_signature():
+    payload = "Please take a moment to register today"
+    keys = [RSAKey(key=import_rsa_key_from_file(KEY))]
+    # keys[0]._keytype = "private"
+    _jws = JWS(payload, alg="RS256")
+    _jwt = _jws.sign_compact(keys)
+
+    p = _jwt.split('.')
+    _jwt = '.'.join(p[:-1])
+
+    _rj = JWS()
+    try:
+        _ = _rj.verify_compact(_jwt, keys)
+    except jwkest.WrongNumberOfParts:
+        pass
+    else:
+        assert False
+
+
 if __name__ == "__main__":
-    test_dj_usage()
+    test_rs256_rm_signature()