Make all import of keys to a KeyBundle go through do_keys().

Author: rohe

src/cryptojwt/jws/utils.py

@@ -45,7 +45,7 @@ def alg2keytype(alg):
     elif alg.startswith("RS") or alg.startswith("PS"):
         return "RSA"
     elif alg.startswith("HS") or alg.startswith("A"):
-        return "oct"
+        return "OCT"
     elif alg.startswith("ES") or alg.startswith("ECDH-ES"):
         return "EC"
     else:

src/cryptojwt/key_bundle.py

@@ -12,6 +12,8 @@
 from .exception import DeSerializationNotPossible
 from .exception import JWKException
 from .exception import UnknownKeyType
+from .exception import UnsupportedAlgorithm
+from .exception import UnsupportedECurve
 from .exception import UpdateFailed
 from .jwk.ec import ECKey
 from .jwk.ec import import_private_key_from_file
@@ -36,11 +38,11 @@
 #     _err = json.dumps({'error': error, 'error_description': descr})
 #     raise excep(_err, 'application/json')
 
-
+# Make sure the keys are all uppercase
 K2C = {
     "RSA": RSAKey,
     "EC": ECKey,
-    "oct": SYMKey,
+    "OCT": SYMKey,
 }
 
 MAP = {'dec': 'enc', 'enc': 'enc', 'ver': 'sig', 'sig': 'sig'}
@@ -242,33 +244,39 @@ def do_keys(self, keys):
         :return:
         """
         for inst in keys:
-            typ = inst["kty"]
+            inst['kty'] = inst["kty"].upper()
+            _typ = inst['kty']
             try:
                 _usage = harmonize_usage(inst['use'])
             except KeyError:
                 _usage = ['']
             else:
                 del inst['use']
 
-            flag = 0
+            _error = ''
             for _use in _usage:
-                for _typ in [typ, typ.lower(), typ.upper()]:
-                    try:
-                        _key = K2C[_typ](use=_use, **inst)
-                    except KeyError:
-                        continue
-                    except JWKException as err:
-                        LOGGER.warning('While loading keys: %s', err)
-                    else:
-                        if _key not in self._keys:
-                            if not _key.kid:
-                                _key.add_kid()
-                            self._keys.append(_key)
-                        flag = 1
-                        break
-            if not flag:
-                LOGGER.warning(
-                    'While loading keys, UnknownKeyType: %s', typ)
+                try:
+                    _key = K2C[_typ](use=_use, **inst)
+                except KeyError:
+                    _error = 'UnknownKeyType: {}'.format(_typ)
+                    continue
+                except (UnsupportedECurve, UnsupportedAlgorithm) as err:
+                    _error = str(err)
+                    break
+                except JWKException as err:
+                    LOGGER.warning('While loading keys: %s', err)
+                    _error = str(err)
+                else:
+                    if _key not in self._keys:
+                        if not _key.kid:
+                            _key.add_kid()
+                        self._keys.append(_key)
+                    _error = ''
+                    break
+            if _error:
+                LOGGER.warning('While loading keys, %s', _error)
+
+        self.last_updated = time.time()
 
     def do_local_jwk(self, filename):
         """
@@ -282,8 +290,6 @@ def do_local_jwk(self, filename):
         else:
             self.do_keys([_info])
 
-        self.last_updated = time.time()
-
     def do_local_der(self, filename, keytype, keyusage=None, kid=''):
         """
         Load a DER encoded file amd create a key from it.
@@ -292,29 +298,25 @@ def do_local_der(self, filename, keytype, keyusage=None, kid=''):
         :param keytype: Presently 'rsa' and 'ec' supported
         :param keyusage: encryption ('enc') or signing ('sig') or both
         """
-        if keytype.lower() == 'rsa':
-            _bkey = import_private_rsa_key_from_file(filename)
-            _key = RSAKey().load_key(_bkey)
-        elif keytype.lower() == 'ec':
-            _bkey = import_private_key_from_file(filename)
-            _key = ECKey().load_key(_bkey)
+        key_args = {}
+        _kty = keytype.lower()
+        if _kty in ['rsa', 'ec']:
+            key_args["kty"] = _kty
+            _key = import_private_rsa_key_from_file(filename)
+            key_args["priv_key"] = _key
+            key_args["pub_key"] = _key.public_key()
         else:
-            raise NotImplementedError('No support for DER decoding of that key type')
+            raise NotImplementedError('No support for DER decoding of key type {}'.format(_kty))
 
         if not keyusage:
-            keyusage = ["enc", "sig"]
+            key_args["use"] = ["enc", "sig"]
         else:
-            keyusage = harmonize_usage(keyusage)
+            key_args["use"] = harmonize_usage(keyusage)
 
-        for use in keyusage:
-            _key.use = use
-            if kid:
-                _key.kid = kid
-            if not _key.kid:
-                _key.add_kid()
-            self._keys.append(_key)
+        if kid:
+            key_args['kid'] = kid
 
-        self.last_updated = time.time()
+        self.do_keys([key_args])
 
     def do_remote(self):
         """

tests/test_03_key_bundle.py

@@ -430,18 +430,18 @@ def test_outdated():
 
 
 def test_dump_jwks():
-    kb1 = rsa_init(
-        {'use': ['enc', 'sig'], 'size': 1024, 'name': 'rsa', 'path': 'keys'})
     a = {"kty": "oct", "key": "highestsupersecret", "use": "sig"}
     b = {"kty": "oct", "key": "highestsupersecret", "use": "enc"}
     kb2 = KeyBundle([a, b])
+
+    kb1 = rsa_init({'use': ['enc', 'sig'], 'size': 1024, 'name': 'rsa', 'path': 'keys'})
     dump_jwks([kb1, kb2], 'jwks_combo')
 
     # Now read it
 
     nkb = KeyBundle(source='file://jwks_combo', fileformat='jwks')
 
-    assert len(nkb) == 2
+    assert len(nkb) == 4
     # both RSA keys
     assert len(nkb.get('rsa')) == 2
 
@@ -656,10 +656,8 @@ def test_keys():
 
 EXPECTED = [
     b'iA7PvG_DfJIeeqQcuXFmvUGjqBkda8In_uMpZrcodVA',
-    b'kLsuyGef1kfw5-t-N9CJLIHx_dpZ79-KemwqjwdrvTI',
-    b'8w34j9PLyCVC7VOZZb1tFVf0MOa2KZoy87lICMeD5w8',
-    b'nKzalL5pJOtVAdCtBAU8giNRNimE-XbylWZ4vq6ZlF8',
-    b'akXzyGlXg8yLhsCczKb_r8VERLx7-iZBUMIVgg2K7p4'
+    b'akXzyGlXg8yLhsCczKb_r8VERLx7-iZBUMIVgg2K7p4',
+    b'Rdy8n5h0fo2q9USHJ6HQKnNZFynN1pWN_X6Bc_Tx-lE'
 ]
 
 

tests/test_06_jws.py

@@ -225,7 +225,7 @@ def test_hmac_512():
 
 def test_hmac_from_keyrep():
     payload = "Please take a moment to register today"
-    symkeys = [k for k in SIGJWKS if k.kty == "oct"]
+    symkeys = [k for k in SIGJWKS if k.kty == "OCT"]
     _jws = JWS(payload, alg="HS512")
     _jwt = _jws.sign_compact(symkeys)