More about making sure the crypto algorithms use expect are the ones used.

Author: rohe

src/cryptojwt/__init__.py

@@ -15,7 +15,7 @@
 except ImportError:
     pass
 
-__version__ = '0.5.0'
+__version__ = '0.6.1'
 
 logger = logging.getLogger(__name__)
 

src/cryptojwt/jwe/jwe.py

@@ -204,8 +204,12 @@ def alg2keytype(self, alg):
         return alg2keytype(alg)
 
 
-def factory(token, **kwargs):
-    _jwt = JWEnc().unpack(token, **kwargs)
+def factory(token, alg='', enc=''):
+    try:
+        _jwt = JWEnc().unpack(token, alg=alg, enc=enc)
+    except KeyError:
+        return None
+
     if _jwt.is_jwe():
         _jwe = JWE()
         _jwe.jwt = _jwt

src/cryptojwt/jws/jws.py

@@ -71,6 +71,11 @@ def valid(self):
 
 
 class JWS(JWx):
+    def __init__(self, msg=None, with_digest=False, httpc=None, **kwargs):
+        JWx.__init__(self, msg, with_digest, httpc, **kwargs)
+        if 'alg' not in self:
+            self['alg'] = "RS256"
+
     def alg_keys(self, keys, use, protected=None):
         _alg = self._pick_alg(keys)
 
@@ -188,9 +193,16 @@ def verify_compact_verbose(self, jws=None, keys=None, allow_none=False,
                 else:
                     raise SignerAlgError("none not allowed")
 
-        if "alg" in self and _alg:
-            if self["alg"] != _alg:
-                raise SignerAlgError("Wrong signing algorithm")
+        if "alg" in self and self['alg'] and _alg:
+            if isinstance(self['alg'], list):
+                if _alg not in self["alg"] :
+                    raise SignerAlgError(
+                        "Wrong signing algorithm, expected {} go {}".format(
+                            self['alg'], _alg))
+            elif _alg != self['alg']:
+                raise SignerAlgError(
+                    "Wrong signing algorithm, expected {} go {}".format(
+                        self['alg'], _alg))
 
         if sigalg and sigalg != _alg:
             raise SignerAlgError("Expected {0} got {1}".format(
@@ -419,15 +431,16 @@ def verify_alg(self, alg):
             return False
 
 
-def factory(token):
+def factory(token, alg=''):
     """
     Instantiate an JWS instance if the token is a signed JWT.
 
     :param token: The token that might be a signed JWT
+    :param alg: The expected signature algorithm
     :return: A JWS instance if the token was a signed JWT, otherwise None
     """
 
-    _jw = JWS()
+    _jw = JWS(alg=alg)
     if _jw.is_jws(token):
         return _jw
     else:

src/cryptojwt/jwt.py

@@ -280,18 +280,17 @@ def unpack(self, token):
         _jwe_header = _jws_header = None
 
         # Check if it's an encrypted JWT
-        _decryptor = jwe_factory(token)
-        if _decryptor:
-            # check headers
-            darg = {}
-            if self.allowed_enc_encs:
-                darg['enc'] = self.allowed_enc_encs
-            if self.allowed_enc_algs:
-                darg['alg'] = self.allowed_enc_algs
-
-            if _decryptor.jwt.verify_headers(**darg) is False:
-                raise HeaderError('Wrong alg or enc')
+        darg = {}
+        if self.allowed_enc_encs:
+            darg['enc'] = self.allowed_enc_encs
+        if self.allowed_enc_algs:
+            darg['alg'] = self.allowed_enc_algs
+        try:
+            _decryptor = jwe_factory(token, **darg)
+        except (KeyError, HeaderError):
+            _decryptor = None
 
+        if _decryptor:
             # Yes, try to decode
             _info = self._decrypt(_decryptor, token)
             _jwe_header = _decryptor.jwt.headers
@@ -307,14 +306,12 @@ def unpack(self, token):
         # If I have reason to believe the information I have is a signed JWT
         if _content_type.lower() == 'jwt':
             # Check that is a signed JWT
-            _verifier = jws_factory(_info)
+            if self.allowed_sign_algs:
+                _verifier = jws_factory(_info, alg=self.allowed_sign_algs)
+            else:
+                _verifier = jws_factory(_info, alg=self.allowed_sign_algs)
+
             if _verifier:
-                if self.allowed_sign_algs and not _verifier.jwt.verify_headers(
-                        alg=self.allowed_sign_algs):
-                    raise HeaderError(
-                        'Wrong signing algorithm: "{}" expected "{}"'.format(
-                            _verifier.jwt.headers['alg'],
-                            self.allowed_sign_algs))
                 _info = self._verify(_verifier, _info)
             else:
                 raise Exception()

src/cryptojwt/simple_jwt.py

@@ -44,6 +44,9 @@ def unpack(self, token, **kwargs):
         self.part = [b64d(p) for p in part]
         self.headers = json.loads(as_unicode(self.part[0]))
         for key,val in kwargs.items():
+            if not val and key in self.headers:
+                continue
+
             try:
                 _ok = self.verify_header(key,val)
             except KeyError:

tests/test_06_jws.py

@@ -165,7 +165,7 @@ def test_1():
                 "exp": 1300819380,
                 "http://example.com/is_root": True}
 
-    _jws = JWS(claimset, cty="JWT")
+    _jws = JWS(claimset, cty="JWT", alg='none')
     _jwt = _jws.sign_compact()
 
     _jr = JWS()
@@ -181,7 +181,7 @@ def test_hmac_256():
     _jws = JWS(payload, alg="HS256")
     _jwt = _jws.sign_compact(keys)
 
-    info = JWS().verify_compact(_jwt, keys)
+    info = JWS(alg="HS256").verify_compact(_jwt, keys)
 
     assert info == payload
 
@@ -192,7 +192,7 @@ def test_hmac_384():
     _jws = JWS(payload, alg="HS384")
     _jwt = _jws.sign_compact(keys)
 
-    _rj = JWS()
+    _rj = JWS(alg="HS384")
     info = _rj.verify_compact(_jwt, keys)
 
     assert info == payload
@@ -204,7 +204,7 @@ def test_hmac_512():
     _jws = JWS(payload, alg="HS512")
     _jwt = _jws.sign_compact(keys)
 
-    _rj = JWS()
+    _rj = JWS(alg="HS512")
     info = _rj.verify_compact(_jwt, keys)
     assert info == payload
 
@@ -215,7 +215,7 @@ def test_hmac_from_keyrep():
     _jws = JWS(payload, alg="HS512")
     _jwt = _jws.sign_compact(symkeys)
 
-    _rj = JWS()
+    _rj = JWS(alg="HS512")
     info = _rj.verify_compact(_jwt, symkeys)
     assert info == payload
 
@@ -239,7 +239,7 @@ def test_rs256():
     _jwt = _jws.sign_compact(skeys)
 
     vkeys = [RSAKey(pub_key=_pkey.public_key())]
-    _rj = JWS()
+    _rj = JWS(alg="RS256")
     info = _rj.verify_compact(_jwt, vkeys)
 
     assert info == payload
@@ -254,7 +254,7 @@ def test_rs384():
     _jwt = _jws.sign_compact(keys)
 
     vkeys = [RSAKey(pub_key=_pkey.public_key())]
-    _rj = JWS()
+    _rj = JWS(alg="RS384")
     info = _rj.verify_compact(_jwt, vkeys)
     assert info == payload
 
@@ -268,7 +268,7 @@ def test_rs512():
     _jwt = _jws.sign_compact(keys)
 
     vkeys = [RSAKey(pub_key=_pkey.public_key())]
-    _rj = JWS()
+    _rj = JWS(alg="RS512")
     info = _rj.verify_compact(_jwt, vkeys)
     assert info == payload
 
@@ -304,8 +304,8 @@ def test_a_1_3a():
             "HAiOjEzMDA4MTkzODAsDQogImh0dHA6Ly9leGFtcGxlLmNvbS9pc19yb290Ijp0cnV"
             "lfQ.dBjftJeZ4CVP-mB92K27uhbUJU1p1r_wW1gFWFOEjXk")
 
-    # keycol = {"hmac": cryptojwt.intarr2bin(HMAC_KEY)}
-    jwt = JWSig().unpack(_jwt)
+    # alg == '' means I'm fine with whatever I get
+    jwt = JWSig(alg='').unpack(_jwt)
     assert jwt.valid()
 
     hmac = intarr2bin(HMAC_KEY)
@@ -318,7 +318,7 @@ def test_a_1_3b():
             "eHAiOjEzMDA4MTkzODAsDQogImh0dHA6Ly9leGFtcGxlLmNvbS9pc19yb290Ijp0c"
             "nVlfQ.dBjftJeZ4CVP-mB92K27uhbUJU1p1r_wW1gFWFOEjXk")
     keys = [SYMKey(key=intarr2bin(HMAC_KEY))]
-    _jws2 = JWS()
+    _jws2 = JWS(alg='')
     _jws2.verify_compact(_jwt, keys)
 
 
@@ -372,7 +372,7 @@ def test_signer_es(ec_func, alg):
     _jwt = _jws.sign_compact(keys)
 
     _pubkey = ECKey().load_key(eck.public_key())
-    _rj = JWS()
+    _rj = JWS(alg=alg)
     info = _rj.verify_compact(_jwt, [_pubkey])
     assert info == payload
 
@@ -386,7 +386,7 @@ def test_signer_es256_verbose():
     _jwt = _jws.sign_compact(keys)
 
     _pubkey = ECKey().load_key(eck.public_key())
-    _rj = JWS()
+    _rj = JWS(alg="ES256")
     info = _rj.verify_compact_verbose(_jwt, [_pubkey])
     assert info['msg'] == payload
     assert info['key'] == _pubkey
@@ -401,7 +401,7 @@ def test_signer_ps256():
     _jwt = _jws.sign_compact(keys)
 
     vkeys = [RSAKey(pub_key=_pkey.public_key())]
-    _rj = JWS()
+    _rj = JWS(alg="PS256")
     info = _rj.verify_compact(_jwt, vkeys)
     assert info == payload
 
@@ -415,7 +415,7 @@ def test_signer_ps256_fail():
     _jwt = _jws.sign_compact(keys)[:-5] + 'abcde'
 
     vkeys = [RSAKey(pub_key=_pkey.public_key())]
-    _rj = JWS()
+    _rj = JWS(alg="PS256")
     try:
         _rj.verify_compact(_jwt, vkeys)
     except BadSignature:
@@ -433,7 +433,7 @@ def test_signer_ps384():
     _jwt = _jws.sign_compact(keys)
 
     vkeys = [RSAKey(pub_key=_pkey.public_key())]
-    _rj = JWS()
+    _rj = JWS(alg="PS384")
     info = _rj.verify_compact(_jwt, vkeys)
     assert info == payload
 
@@ -448,7 +448,7 @@ def test_signer_ps512():
     _jwt = _jws.sign_compact(keys)
 
     vkeys = [RSAKey(pub_key=_pkey.public_key())]
-    _rj = factory(_jwt)
+    _rj = factory(_jwt, alg="PS512")
     info = _rj.verify_compact(_jwt, vkeys)
     assert info == payload
     assert _rj.verify_alg('PS512')
@@ -462,7 +462,7 @@ def test_no_alg_and_alg_none_same():
     _jwt0 = _jws.sign_compact([])
 
     # The class instance that sets up the signing operation
-    _jws = JWS(payload)
+    _jws = JWS(payload, alg="none")
 
     # Create a JWS (signed JWT)
     _jwt1 = _jws.sign_compact([])
@@ -506,7 +506,7 @@ def test_signer_protected_headers():
     assert b64d(enc_payload.encode("utf-8")).decode("utf-8") == payload
 
     _pub_key = ECKey().load_key(eck.public_key())
-    _rj = JWS()
+    _rj = JWS(alg='ES256')
     info = _rj.verify_compact(_jwt, [_pub_key])
     assert info == payload
 
@@ -699,7 +699,7 @@ def test_pick_alg_assume_alg_from_single_key():
     expected_alg = "HS256"
     keys = [SYMKey(key="foobar subdued thought", alg=expected_alg)]
 
-    alg = JWS()._pick_alg(keys)
+    alg = JWS(alg=expected_alg)._pick_alg(keys)
     assert alg == expected_alg
 
 

tests/test_07_jwe.py

@@ -276,7 +276,7 @@ def test_encrypt_decrypt_rsa_cbc():
 
     jwt = _jwe0.encrypt([_key])
 
-    _jwe1 = factory(jwt)
+    _jwe1 = factory(jwt, alg="RSA1_5", enc="A128CBC-HS256")
     _dkey = RSAKey(priv_key=priv_key)
     _dkey._keytype = "private"
     msg = _jwe1.decrypt(jwt, [_dkey])
@@ -318,7 +318,7 @@ def test_ecdh_encrypt_decrypt_direct_key():
     jwt = jwenc.encrypt(**kwargs)
 
     # Bob decrypts
-    ret_jwe = factory(jwt)
+    ret_jwe = factory(jwt, alg="ECDH-ES", enc="A128GCM")
     jwdec = JWE_EC()
     jwdec.dec_setup(ret_jwe.jwt, key=bob)
     msg = jwdec.decrypt(ret_jwe.jwt)
@@ -342,7 +342,7 @@ def test_ecdh_encrypt_decrypt_keywrapped_key():
 
     jwt = jwenc.encrypt(**kwargs)
 
-    ret_jwe = factory(jwt)
+    ret_jwe = factory(jwt, alg="ECDH-ES+A128KW", enc="A128GCM")
     jwdec = JWE_EC()
     jwdec.dec_setup(ret_jwe.jwt, key=bob)
     msg = jwdec.decrypt(ret_jwe.jwt)
@@ -365,7 +365,7 @@ def test_ecdh_no_setup_dynamic_epk():
     jwenc = JWE(plain, alg="ECDH-ES", enc="A128GCM")
     jwt = jwenc.encrypt([eck_bob])
     assert jwt
-    ret_jwe = factory(jwt)
+    ret_jwe = factory(jwt, alg="ECDH-ES", enc="A128GCM")
     res = ret_jwe.decrypt(jwt, [eck_bob])
     assert res == plain
 
@@ -374,7 +374,7 @@ def test_verify_headers():
     jwenc = JWE(plain, alg="ECDH-ES", enc="A128GCM")
     jwt = jwenc.encrypt([eck_bob])
     assert jwt
-    decryptor = factory(jwt)
+    decryptor = factory(jwt, alg="ECDH-ES", enc="A128GCM")
     assert decryptor.jwt.verify_headers(alg='ECDH-ES', enc='A128GCM')
     assert decryptor.jwt.verify_headers(alg='RS256') is False
     assert decryptor.jwt.verify_headers(kid='RS256') is False

tests/test_09_jwt.py

@@ -52,11 +52,11 @@ def test_jwt_pack():
 
 
 def test_jwt_pack_and_unpack():
-    alice = JWT(key_jar=ALICE_KEY_JAR, iss=ALICE)
+    alice = JWT(key_jar=ALICE_KEY_JAR, iss=ALICE, sign_alg='RS256')
     payload = {'sub': 'sub'}
     _jwt = alice.pack(payload=payload)
 
-    bob = JWT(key_jar=BOB_KEY_JAR, iss=BOB)
+    bob = JWT(key_jar=BOB_KEY_JAR, iss=BOB, allowed_sign_algs=["RS256"])
     info = bob.unpack(_jwt)
 
     assert set(info.keys()) == {'iat', 'iss', 'sub', 'kid'}