Stricter treatment of JWT headers.

Signature algorithm should always be checked, the same for
encryption alg and enc.

Author: rohe

src/cryptojwt/jwe/jwe.py

@@ -61,8 +61,6 @@ def encrypt(self, keys=None, cek="", iv="", **kwargs):
         """
 
         :param keys: A set of possibly usable keys
-        :param context: If the other party's public or my private key should be
-            used for encryption
         :param cek: Content master key
         :param iv: Initialization vector
         :param kwargs: Extra key word arguments
@@ -206,8 +204,8 @@ def alg2keytype(self, alg):
         return alg2keytype(alg)
 
 
-def factory(token):
-    _jwt = JWEnc().unpack(token)
+def factory(token, **kwargs):
+    _jwt = JWEnc().unpack(token, **kwargs)
     if _jwt.is_jwe():
         _jwe = JWE()
         _jwe.jwt = _jwt

src/cryptojwt/jws/jws.py

@@ -410,33 +410,24 @@ def verify_alg(self, alg):
         Specifically check that the 'alg' claim has a specific value
 
         :param alg: The expected alg value
-        :raises: KeyError if the 'alg' is not present in the header
         :return: True if the alg value in the header is the same as the one
-            given.
+            given. Returns False if no 'alg' claim exists in the header.
         """
-        if alg == self.jwt.headers['alg']:
-            return True
-        else:
+        try:
+            return self.jwt.verify_header('alg', alg)
+        except KeyError:
             return False
 
-    def verify_header(self, key, val):
-        """
-        Check that a particular header claim is present as a has specific value
 
-        :param key: The claim
-        :param val: The value of the claim
-        :raises: KeyError if the claim is not present in the header
-        :return: True if the claim exists in the header and has the prescribed
-            value
-        """
-        if val == self.jwt.headers[key]:
-            return True
-        else:
-            return False
+def factory(token):
+    """
+    Instantiate an JWS instance if the token is a signed JWT.
 
+    :param token: The token that might be a signed JWT
+    :return: A JWS instance if the token was a signed JWT, otherwise None
+    """
 
-def factory(token, **kwargs):
-    _jw = JWS(**kwargs)
+    _jw = JWS()
     if _jw.is_jws(token):
         return _jw
     else:

src/cryptojwt/jwt.py

@@ -5,6 +5,7 @@
 from json import JSONDecodeError
 
 
+from .exception import HeaderError
 from .exception import VerificationError
 from .utils import as_unicode
 from .jwe.utils import alg2keytype as jwe_alg2keytype
@@ -69,7 +70,7 @@ def __init__(self, key_jar=None, iss='', lifetime=0,
         self.iss = iss  # My identifier
         self.lifetime = lifetime  # default life time of the signature
         self.sign = sign  # default signing or not
-        self.sign_alg = sign_alg  # default signing algorithm
+        self.alg = sign_alg  # default signing algorithm
         self.encrypt = encrypt  # default encrypting or not
         self.enc_alg = enc_alg  # CEK encryption algorithm
         self.enc_enc = enc_enc  # content encryption algorithm
@@ -118,11 +119,11 @@ def put_together_aud(recv, aud=None):
         :return: A possibly extended audience set
         """
         if aud:
-            if recv in aud:
-                _aud = aud
-            elif recv:
+            if recv and recv not in aud:
                 _aud = [recv]
                 _aud.extend(aud)
+            else:
+                _aud = aud
         elif recv:
             _aud = [recv]
         else:
@@ -154,7 +155,7 @@ def pack_key(self, owner_id='', kid=''):
         :param kid: Key ID
         :return: One key
         """
-        keys = pick_key(self.my_keys(owner_id, 'sig'), 'sig', alg=self.sign_alg,
+        keys = pick_key(self.my_keys(owner_id, 'sig'), 'sig', alg=self.alg,
                         kid=kid)
 
         if not keys:
@@ -198,13 +199,13 @@ def pack(self, payload=None, kid='', owner='', recv='', aud=None, **kwargs):
             owner = self.iss
 
         if self.sign:
-            if self.sign_alg != 'none':
+            if self.alg != 'none':
                 _key = self.pack_key(owner, kid)
                 _args['kid'] = _key.kid
             else:
                 _key = None
 
-            _jws = JWS(json.dumps(_args), alg=self.sign_alg)
+            _jws = JWS(json.dumps(_args), alg=self.alg)
             _sjwt = _jws.sign_compact([_key])
         else:
             _sjwt = json.dumps(_args)
@@ -242,7 +243,8 @@ def _decrypt(self, rj, token):
             keys = self.key_jar.get_jwt_decrypt_keys(rj.jwt)
         return rj.decrypt(token, keys=keys)
 
-    def verify_profile(self, msg_cls, info, **kwargs):
+    @staticmethod
+    def verify_profile(msg_cls, info, **kwargs):
         """
         If a message type is known for this JSON document. Verify that the
         document complies with the message specifications.
@@ -273,14 +275,24 @@ def unpack(self, token):
         _jwe_header = _jws_header = None
 
         # Check if it's an encrypted JWT
-        _rj = jwe_factory(token)
-        if _rj:
+        _decryptor = jwe_factory(token)
+        if _decryptor:
+            # check headers
+            darg = {}
+            if self.enc_enc:
+                darg['enc'] = self.enc_enc
+            if self.enc_alg:
+                darg['alg'] = self.enc_alg
+
+            if _decryptor.jwt.verify_headers(**darg) is False:
+                raise HeaderError('Wrong alg or enc')
+
             # Yes, try to decode
-            _info = self._decrypt(_rj, token)
-            _jwe_header = _rj.jwt.headers
+            _info = self._decrypt(_decryptor, token)
+            _jwe_header = _decryptor.jwt.headers
             # Try to find out if the information encrypted was a signed JWT
             try:
-                _content_type = _rj.jwt.headers['cty']
+                _content_type = _decryptor.jwt.headers['cty']
             except KeyError:
                 pass
         else:
@@ -289,12 +301,16 @@ 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
-            _rj = jws_factory(_info)
-            if _rj:
-                _info = self._verify(_rj, _info)
+            _verifier = jws_factory(_info)
+            if _verifier:
+                if self.alg and not _verifier.jwt.verify_headers(alg=self.alg):
+                    raise HeaderError(
+                        'Wrong signing algorithm: "{}" expected "{}"'.format(
+                            _verifier.jwt.headers['alg'], self.alg))
+                _info = self._verify(_verifier, _info)
             else:
                 raise Exception()
-            _jws_header = _rj.jwt.headers
+            _jws_header = _verifier.jwt.headers
         else:
             # So, not a signed JWT
             try:

src/cryptojwt/simple_jwt.py

@@ -1,6 +1,7 @@
 import json
 import logging
 
+from cryptojwt.exception import HeaderError
 from .utils import as_unicode
 from .utils import b64d
 from .utils import b64encode_item
@@ -23,12 +24,14 @@ def __init__(self, **headers):
         self.b64part = [b64encode_item(headers)]
         self.part = [b64d(self.b64part[0])]
 
-    def unpack(self, token):
+    def unpack(self, token, **kwargs):
         """
         Unpacks a JWT into its parts and base64 decodes the parts
         individually
 
         :param token: The JWT
+        :param kwargs: A possible empty set of claims to verify the header
+            against.
         """
         if isinstance(token, str):
             try:
@@ -39,8 +42,18 @@ def unpack(self, token):
         part = split_token(token)
         self.b64part = part
         self.part = [b64d(p) for p in part]
-        #self.headers = json.loads(self.part[0].decode())
         self.headers = json.loads(as_unicode(self.part[0]))
+        for key,val in kwargs.items():
+            try:
+                _ok = self.verify_header(key,val)
+            except KeyError:
+                raise
+            else:
+                if not _ok:
+                    raise HeaderError(
+                        'Expected "{}" to be "{}", was "{}"'.format(
+                            key, val, self.headers[key]))
+
         return self
 
     def pack(self, parts=None, headers=None):
@@ -87,4 +100,43 @@ def payload(self):
             except ValueError:
                 pass
 
-        return _msg
+        return _msg
+
+    def verify_header(self, key, val):
+        """
+        Check that a particular header claim is present and has a specific value
+
+        :param key: The claim
+        :param val: The value of the claim
+        :raises: KeyError if the claim is not present in the header
+        :return: True if the claim exists in the header and has the prescribed
+            value
+        """
+
+        if self.headers[key] == val:
+            return True
+        else:
+            return False
+
+    def verify_headers(self, check_presence=True, **kwargs):
+        """
+        Check that a set of particular header claim are present and has
+        specific values
+
+        :param kwargs: The claim/value sets as a dictionary
+        :return: True if the claim that appears in the header has the
+            prescribed values. If a claim is not present in the header and
+            check_presence is True then False is returned.
+        """
+        for key, val in kwargs.items():
+            try:
+                _ok = self.verify_header(key, val)
+            except KeyError:
+                if check_presence:
+                    return False
+                else:
+                    pass
+            else:
+                if not _ok:
+                    return False
+        return True

tests/test_06_jws.py

@@ -828,6 +828,5 @@ def test_factory_verify_alg():
     _signer = JWS(payload, alg='RS256')
     _signer.set_header_claim('foo', 'bar')
     _jws = _signer.sign_compact(keys, abc=123)
-    _verifier = factory(_jws, alg='RS512')
-    with pytest.raises(SignerAlgError):
-        _verifier.verify_compact(_jws, keys)
+    _verifier = factory(_jws)
+    assert _verifier.jwt.verify_headers(alg='RS512') is False

tests/test_07_jwe.py

@@ -368,3 +368,13 @@ def test_ecdh_no_setup_dynamic_epk():
     ret_jwe = factory(jwt)
     res = ret_jwe.decrypt(jwt, [eck_bob])
     assert res == plain
+
+
+def test_verify_headers():
+    jwenc = JWE(plain, alg="ECDH-ES", enc="A128GCM")
+    jwt = jwenc.encrypt([eck_bob])
+    assert jwt
+    decryptor = factory(jwt)
+    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

@@ -93,7 +93,7 @@ def test_jwt_pack_unpack_sym():
 
     _kj = KeyJar()
     _kj.add_symmetric(ALICE, 'hemligt ordsprak', usage=['sig'])
-    bob = JWT(key_jar=_kj, iss=BOB)
+    bob = JWT(key_jar=_kj, iss=BOB, sign_alg="HS256")
     info = bob.unpack(_jwt)
     assert info
 
@@ -115,7 +115,7 @@ def test_jwt_pack_and_unpack_with_alg():
     payload = {'sub': 'sub'}
     _jwt = alice.pack(payload=payload)
 
-    bob = JWT(BOB_KEY_JAR)
+    bob = JWT(BOB_KEY_JAR, sign_alg='RS384')
     info = bob.unpack(_jwt)
 
     assert set(info.keys()) == {'iat', 'iss', 'sub', 'kid'}
@@ -138,7 +138,7 @@ def test_with_jti():
     payload = {'sub': 'sub2'}
     _jwt = alice.pack(payload=payload)
 
-    bob = JWT(key_jar=_kj, iss=BOB)
+    bob = JWT(key_jar=_kj, iss=BOB, sign_alg="HS256")
     info = bob.unpack(_jwt)
     assert 'jti' in info
 
@@ -160,7 +160,7 @@ def test_msg_cls():
     payload = {'sub': 'sub2'}
     _jwt = alice.pack(payload=payload)
 
-    bob = JWT(key_jar=_kj, iss=BOB)
+    bob = JWT(key_jar=_kj, iss=BOB, sign_alg="HS256")
     bob.msg_cls = DummyMsg
     info = bob.unpack(_jwt)
     assert isinstance(info, DummyMsg)