Using teh jwk claim in a JWT for signing / verifying signature.

Author: rohe

src/cryptojwt/jwe/jwe.py

@@ -1,5 +1,6 @@
 import logging
 
+from cryptojwt.jwk.jwk import key_from_jwk_dict
 from ..jwk.asym import AsymmetricKey
 from ..jwx import JWx
 
@@ -147,6 +148,11 @@ def decrypt(self, token=None, keys=None, alg=None, cek=None):
         else:
             keys = self.pick_keys(self._get_keys(), use="enc", alg=_alg)
 
+        try:
+            keys.append(key_from_jwk_dict(_jwe.headers['jwk']))
+        except KeyError:
+            pass
+
         if not keys and not cek:
             raise NoSuitableDecryptionKey(_alg)
 

src/cryptojwt/jwe/jwe_ec.py

@@ -177,9 +177,18 @@ def dec_setup(self, token, key=None, **kwargs):
 
         return self.cek
 
-    def encrypt(self, iv="", cek="", **kwargs):
+    def encrypt(self, key=None, iv="", cek="", **kwargs):
+        """
 
+        :param key: *Not used>, only there to present the same API as
+            JWE_RSA and JWE_SYM
+        :param iv:
+        :param cek:
+        :param kwargs:
+        :return:
+        """
         _msg = as_bytes(self.msg)
+
         _args = self._dict
         try:
             _args["kid"] = kwargs["kid"]

src/cryptojwt/jwk/__init__.py

@@ -44,10 +44,11 @@ def __init__(self, kty="", alg="", use="", kid="", x5c=None,
             if not isinstance(alg, str):
                 alg = as_unicode(alg)
 
-            if alg not in [
-                "HS256", "HS384", "HS512", "RS256", "RS384", "RS512", "ES256",
-                "ES384","ES512", "PS256", "PS384", "PS512"
-                ]:
+            # The list comes from https://tools.ietf.org/html/rfc7518#page-6
+            # Should map against SIGNER_ALGS in cryptojwt.jws.jws
+            if alg not in ["HS256", "HS384", "HS512", "RS256", "RS384",
+                           "RS512", "ES256", "ES384","ES512", "PS256",
+                           "PS384", "PS512", "none"]:
                 raise UnsupportedAlgorithm("Unknown algorithm: {}".format(alg))
 
         self.alg = alg
@@ -148,7 +149,7 @@ def verify(self):
 
         if self.kid:
             if not isinstance(self.kid, str):
-                raise HeaderError("kid of wrong value type")
+                raise ValueError("kid of wrong value type")
         return True
 
     def __eq__(self, other):

src/cryptojwt/jwx.py

@@ -2,6 +2,8 @@
 import logging
 
 import requests
+
+from cryptojwt.jwk import JWK
 from cryptojwt.key_bundle import KeyBundle
 
 from .jwk.jwk import key_from_jwk_dict
@@ -67,28 +69,36 @@ def __init__(self, msg=None, with_digest=False, httpc=None, **kwargs):
                     continue
 
                 if key == "jwk":
+                    # value MUST be a string
                     if isinstance(_val, dict):
-                        self._dict["jwk"] = key_from_jwk_dict(_val)
+                        _k = key_from_jwk_dict(_val)
+                        self._dict["jwk"] = _val
                     elif isinstance(_val, str):
-                        self._dict["jwk"] = key_from_jwk_dict(json.loads(_val))
+                        # verify that it's a real JWK
+                        _val = json.loads(_val)
+                        _j = key_from_jwk_dict(_val)
+                        self._dict["jwk"] = _val
+                    elif isinstance(_val, JWK):
+                        self._dict['jwk'] = _val.to_dict()
                     else:
                         raise ValueError(
-                            'JWK must be a string or a JSON object')
+                            'JWK must be a string a JSON object or a JWK '
+                            'instance')
                     self._jwk = self._dict['jwk']
                 elif key == "x5c":
                     self._dict["x5c"] = _val
                     _pub_key = import_rsa_key(_val)
-                    self._jwk = RSAKey(_pub_key)
+                    self._jwk = RSAKey(pub_key=_pub_key).to_dict()
                 elif key == "jku":
                     self._jwks = KeyBundle(source=_val, httpc=self.httpc)
                     self._dict['jku'] = _val
                 elif "x5u" in self:
                     try:
                         _spec = load_x509_cert(self["x5u"], self.httpc, {})
-                        self._jwk = RSAKey(pub_key=_spec['rsa'])
+                        self._jwk = RSAKey(pub_key=_spec['rsa']).to_dict()
                     except Exception:
                         # ca_chain = load_x509_cert_chain(self["x5u"])
-                        pass
+                        raise ValueError('x5u')
                 else:
                     self._dict[key] = _val
 
@@ -110,12 +120,11 @@ def __getattr__(self, item):
     def keys(self):
         return list(self._dict.keys())
 
-    def headers(self, extra=None):
-        _extra = extra or {}
+    def headers(self, **kwargs):
         _header = self._header.copy()
         for param in self.args:
             try:
-                _header[param] = _extra[param]
+                _header[param] = kwargs[param]
             except KeyError:
                 try:
                     if self._dict[param]:
@@ -124,9 +133,27 @@ def headers(self, extra=None):
                     pass
 
         if "jwk" in self:
-            _header["jwk"] = self["jwk"].serialize()
-        elif "jwk" in _extra:
-            _header["jwk"] = extra["jwk"].serialize()
+            _header["jwk"] = self["jwk"]
+        else:
+            try:
+                _jwk = kwargs['jwk']
+            except KeyError:
+                pass
+            else:
+                try:
+                    _header["jwk"] = _jwk.serialize()  # JWK instance
+                except AttributeError:
+                    if isinstance(_jwk, dict):
+                        _header['jwk'] = _jwk  # dictionary
+                    else:
+                        try:
+                            _d = json.loads(_jwk) # JSON
+                            # Verify that it's a valid JWK
+                            _k = key_from_jwk_dict(_d)
+                        except Exception:
+                            raise
+                        else:
+                            _header['jwk'] = _d
 
         if "kid" in self:
             if not isinstance(self["kid"], str):
@@ -135,12 +162,9 @@ def headers(self, extra=None):
         return _header
 
     def _get_keys(self):
-        logger.debug("_get_keys(): self._dict.keys={0}".format(
-            self._dict.keys()))
-
         _keys = []
         if self._jwk:
-            _keys.append(self._jwk)
+            _keys.append(key_from_jwk_dict(self._jwk))
         if self._jwks is not None:
             _keys.extend(self._jwks.keys())
         return _keys
@@ -153,8 +177,8 @@ def pick_keys(self, keys, use="", alg=""):
         The assumption is that upper layer has made certain you only get
         keys you can use.
 
-        :param alg:
-        :param use:
+        :param alg: The crypto algorithm
+        :param use: What the key should be used for
         :param keys: A list of JWK instances
         :return: A list of JWK instances that fulfill the requirements
         """

tests/test_02_jwk.py

@@ -15,14 +15,15 @@
 
 import os.path
 
-from cryptojwt.exception import DeSerializationNotPossible
+from cryptojwt.exception import DeSerializationNotPossible, UnsupportedAlgorithm
 from cryptojwt.exception import WrongUsage
 
-from cryptojwt.utils import as_unicode
+from cryptojwt.utils import as_unicode, as_bytes
 from cryptojwt.utils import b64e
 from cryptojwt.utils import long2intarr
 from cryptojwt.utils import base64url_to_long
 from cryptojwt.utils import base64_to_long
+from cryptojwt.jwk import JWK
 from cryptojwt.jwk.ec import ECKey
 from cryptojwt.jwk.rsa import import_private_rsa_key_from_file
 from cryptojwt.jwk.rsa import import_public_rsa_key_from_file
@@ -50,7 +51,7 @@ def full_path(local_file):
 N = 'wf-wiusGhA-gleZYQAOPQlNUIucPiqXdPVyieDqQbXXOPBe3nuggtVzeq7pVFH1dZz4dY2Q2LA5DaegvP8kRvoSB_87ds3dy3Rfym_GUSc5B0l1TgEobcyaep8jguRoHto6GWHfCfKqoUYZq4N8vh4LLMQwLR6zi6Jtu82nB5k8'
 E = 'AQAB'
 
-JWK = {"keys": [
+JWK_0 = {"keys": [
     {'kty': 'RSA', 'use': 'foo', 'e': E, 'kid': "abc",
      'n': N}
 ]}
@@ -98,8 +99,8 @@ def test_kspec():
 
     jwk = _key.serialize()
     assert jwk["kty"] == "RSA"
-    assert jwk["e"] == JWK["keys"][0]["e"]
-    assert jwk["n"] == JWK["keys"][0]["n"]
+    assert jwk["e"] == JWK_0["keys"][0]["e"]
+    assert jwk["n"] == JWK_0["keys"][0]["n"]
 
     assert not _key.has_private_key()
 
@@ -467,3 +468,54 @@ def test_key_from_jwk_dict_sym():
     assert isinstance(_key, SYMKey)
     jwk = _key.serialize()
     assert jwk == {'kty': 'oct', 'k': 'YWJjZGVmZ2hpamtsbW5vcHE'}
+
+
+def test_jwk_wrong_alg():
+    with pytest.raises(UnsupportedAlgorithm):
+        _j = JWK(alg='xyz')
+
+
+def test_jwk_conversion():
+    _j = JWK(use=b'sig', kid=b'1', alg=b'RS512')
+    assert _j.use == 'sig'
+    args = _j.common()
+    assert set(args.keys()) == {'kty', 'use', 'kid', 'alg'}
+
+
+def test_str():
+    _j = RSAKey(alg="RS512", use='sig', n=N, e=E)
+    s = '{}'.format(_j)
+    assert s.startswith("{") and s.endswith("}")
+    sp = s.replace("'", '"')
+    _d = json.loads(sp)
+    assert set(_d.keys()) == {'alg', 'use', 'n', 'e', 'kty'}
+
+
+def test_verify():
+    _j = RSAKey(alg=b"RS512", use=b'sig', n=as_bytes(N), e=E)
+    assert _j.verify()
+
+
+def test_verify_wrong_kid():
+    _j = RSAKey(alg=b"RS512", use=b'sig', n=as_bytes(N), e=E, kid=1)
+    with pytest.raises(ValueError):
+        _j.verify()
+
+
+def test_cmp():
+    _j1 = RSAKey(alg="RS256", use="sig", n=N, e=E)
+    _j2 = RSAKey(alg="RS256", use="sig", n=N, e=E)
+    assert _j1 == _j2
+
+
+def test_cmp_jwk():
+    _j1 = JWK(use='sig', kid='1', alg='RS512')
+    _j2 = JWK(use='sig', kid='1', alg='RS512')
+
+    assert _j1 == _j2
+
+def test_appropriate():
+    _j1 = JWK(use='sig', kid='1', alg='RS512')
+
+    assert _j1.appropriate_for('sign')
+    assert _j1.appropriate_for('encrypt') is False

tests/test_05_jwx.py

@@ -77,7 +77,7 @@ def test_get_headers():
     _headers = jwx.headers()
     assert set(_headers.keys()) == {'jwk', 'alg'}
 
-    _headers = jwx.headers({'kid': '123'})
+    _headers = jwx.headers(kid='123')
     assert set(_headers.keys()) == {'jwk', 'alg', 'kid'}
 
 
@@ -92,3 +92,9 @@ def test_decode():
     jwx = JWx(cty='JWT')
     _msg = jwx._decode(b'eyJmb28iOiJiYXIifQ')
     assert _msg == {'foo':'bar'}
+
+
+def test_extra_headers():
+    jwx = JWx()
+    headers = jwx.headers(jwk=JSON_RSA_PUB_KEY, alg="RS256")
+    assert set(headers.keys()) == {'jwk', 'alg'}

tests/test_07_jwe.py

@@ -10,7 +10,8 @@
 from cryptojwt.exception import MissingKey
 from cryptojwt.exception import Unsupported
 from cryptojwt.exception import VerificationError
-from cryptojwt.jwe.exception import UnsupportedBitLength
+from cryptojwt.jwe.exception import UnsupportedBitLength, \
+    NoSuitableEncryptionKey
 
 from cryptojwt.utils import b64e
 
@@ -375,4 +376,21 @@ def test_verify_headers():
     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
+    assert decryptor.jwt.verify_headers(kid='RS256') is False
+
+
+def test_encrypt_no_keys():
+    jwenc = JWE(plain, alg="ECDH-ES", enc="A128GCM")
+    with pytest.raises(NoSuitableEncryptionKey):
+        jwenc.encrypt()
+
+
+def test_encrypt_jwk_key():
+    # This is a weird case. Signing the JWT with a key that is
+    # published in the JWT. Still it should be possible.
+    jwenc = JWE(plain, alg="ECDH-ES", enc="A128GCM", jwk=eck_bob)
+    _enc = jwenc.encrypt()
+    assert _enc
+    decryptor = factory(_enc, alg="ECDH-ES", enc="A128GCM")
+    res = decryptor.decrypt()
+    assert res == plain