Fixed a couple of small bugs.

Added 2 new methods: dum&load to KeyBundle and KeyJar.

Author: rohe

src/cryptojwt/jwk/__init__.py

@@ -93,7 +93,7 @@ def __init__(self, kty="", alg="", use="", kid="", x5c=None,
         self.x5c = x5c or []
         self.x5t = x5t
         self.x5u = x5u
-        self.inactive_since = 0
+        self.inactive_since = kwargs.get("inactive_since", 0)
 
     def to_dict(self):
         """

src/cryptojwt/jwk/hmac.py

@@ -140,6 +140,10 @@ def __eq__(self, other):
 
         for key in self.public_members:
             if getattr(other, key) != getattr(self, key):
+                if key == 'kid':
+                    # if one has a value and the other not then assume they are the same
+                    if getattr(self, key) == '' or getattr(other, key) == '':
+                        return True
                 return False
 
         return True

src/cryptojwt/jwk/rsa.py

@@ -14,6 +14,7 @@
 from ..utils import b64e
 from ..utils import deser
 from ..utils import long_to_base64
+from ..utils import as_unicode
 from . import JWK
 from .asym import AsymmetricKey
 
@@ -424,7 +425,7 @@ def serialize(self, private=False):
                 if item:
                     res[param] = item
         if self.x5c:
-            res['x5c'] = [x.decode('utf-8') for x in self.x5c]
+            res['x5c'] = [as_unicode(x) for x in self.x5c]
 
         return res
 

src/cryptojwt/key_bundle.py

@@ -6,17 +6,16 @@
 from functools import cmp_to_key
 
 import requests
+
 from cryptojwt.jwk.ec import NIST2SEC
 from cryptojwt.jwk.hmac import new_sym_key
-
 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
 from .jwk.ec import new_ec_key
 from .jwk.hmac import SYMKey
 from .jwk.jwk import dump_jwk
@@ -649,6 +648,44 @@ def difference(self, bundle):
 
         return [k for k in self._keys if k not in bundle]
 
+    def dump(self):
+        _keys = []
+        for _k in self._keys:
+            _ser = _k.to_dict()
+            if _k.inactive_since:
+                _ser['inactive_since'] = _k.inactive_since
+            _keys.append(_ser)
+
+        res = {
+            "keys": _keys,
+            "fileformat": self.fileformat,
+            "last_updated": self.last_updated,
+            "httpc_params": self.httpc_params,
+            "remote": self.remote,
+            "imp_jwks": self.imp_jwks,
+            "time_out": self.time_out,
+            "cache_time": self.cache_time
+        }
+
+        if self.source:
+            res['source'] = self.source
+
+        return res
+
+    def load(self, spec):
+        _keys = spec.get("keys", [])
+        if _keys:
+            self.do_keys(_keys)
+        self.source = spec.get("source", None)
+        self.fileformat = spec.get("fileformat", "jwks")
+        self.last_updated = spec.get("last_updated", 0)
+        self.remote = spec.get("remote", False)
+        self.imp_jwks = spec.get('imp_jwks', None)
+        self.time_out = spec.get('time_out', 0)
+        self.cache_time = spec.get('cache_time', 0)
+        self.httpc_params = spec.get('httpc_params', {})
+        return self
+
 
 def keybundle_from_local_file(filename, typ, usage, keytype="RSA"):
     """

src/cryptojwt/key_jar.py

@@ -638,6 +638,30 @@ def __len__(self):
                     keys += len(kb)
         return keys
 
+    def dump(self):
+        """
+        Returns the key jar content as dictionary
+
+        :return: A dictionary
+        """
+
+        info = {}
+        for iss in list(self.owners()):
+            info[iss] = []
+            for kb in self.issuer_keys[iss]:
+                info[iss].append(kb.dump())
+        return info
+
+    def load(self, info):
+        """
+
+        :param info: A dictionary with the information
+        :return:
+        """
+        for iss, kbs in info.items():
+            self.issuer_keys[iss] = [KeyBundle().load(val) for val in kbs]
+        return self
+
 
 # =============================================================================
 

tests/test_03_key_bundle.py

@@ -8,8 +8,9 @@
 import requests
 import responses
 from cryptography.hazmat.primitives.asymmetric import rsa
-from cryptojwt.jwk.ec import new_ec_key
+
 from cryptojwt.jwk.ec import ECKey
+from cryptojwt.jwk.ec import new_ec_key
 from cryptojwt.jwk.hmac import SYMKey
 from cryptojwt.jwk.rsa import RSAKey
 from cryptojwt.jwk.rsa import import_rsa_key_from_cert_file
@@ -448,7 +449,7 @@ def test_dump_jwks():
     assert len(nkb.get('rsa')) == 2
 
     # Will dump symmetric keys
-    dump_jwks([kb1, kb2], 'jwks_combo',symmetric_too=True)
+    dump_jwks([kb1, kb2], 'jwks_combo', symmetric_too=True)
 
     # Now read it
     nkb = KeyBundle(source='file://jwks_combo', fileformat='jwks')
@@ -920,3 +921,48 @@ def test_init_key():
     # Now _jwk3 is stored in the file
     _jwk4 = init_key(filename, "RSA")
     assert _jwk4 == _jwk3
+
+
+def test_export_inactive():
+    desc = {"kty": "oct", "key": "highestsupersecret", "use": "sig"}
+    kb = KeyBundle([desc])
+    assert len(kb.keys()) == 1
+    for k in kb.keys():
+        kb.mark_as_inactive(k.kid)
+    desc = {"kty": "oct", "key": "highestsupersecret", "use": "enc"}
+    kb.do_keys([desc])
+    res = kb.dump()
+    assert set(res.keys()) == {'cache_time',
+                               'fileformat',
+                               'httpc_params',
+                               'imp_jwks',
+                               'keys',
+                               'last_updated',
+                               'remote',
+                               'time_out'}
+
+    kb2 = KeyBundle().load(res)
+    assert len(kb2.keys()) == 2
+    assert len(kb2.active_keys()) == 1
+
+
+def test_remote():
+    source = 'https://example.com/keys.json'
+    # Mock response
+    with responses.RequestsMock() as rsps:
+        rsps.add(method="GET", url=source, json=JWKS_DICT, status=200)
+        httpc_params = {'timeout': (2, 2)}  # connect, read timeouts in seconds
+        kb = KeyBundle(source=source, httpc=requests.request,
+                       httpc_params=httpc_params)
+        kb.do_remote()
+
+    exp = kb.dump()
+    kb2 = KeyBundle().load(exp)
+    assert kb2.source == source
+    assert len(kb2.keys()) == 3
+    assert len(kb2.get("rsa")) == 1
+    assert len(kb2.get("oct")) == 1
+    assert len(kb2.get("ec")) == 1
+    assert kb2.httpc_params == {'timeout': (2, 2)}
+    assert kb2.imp_jwks
+    assert kb2.last_updated

tests/test_04_key_jar.py

@@ -4,6 +4,7 @@
 import time
 
 import pytest
+
 from cryptojwt.exception import JWKESTException
 from cryptojwt.jwe.jwenc import JWEnc
 from cryptojwt.jws.jws import JWS
@@ -960,3 +961,17 @@ def test_init_key_jar_create_directories():
     _keyjar = init_key_jar(**OIDC_KEYS)
     assert len(_keyjar.get_signing_key('RSA')) == 1
     assert len(_keyjar.get_signing_key('EC')) == 1
+
+
+def test_dump():
+    kj = KeyJar()
+    kj['Alice'] = [KeyBundle(JWK0['keys'])]
+    kj['Bob'] = [KeyBundle(JWK1['keys'])]
+    kj['C'] = [KeyBundle(JWK2['keys'])]
+
+    res = kj.dump()
+
+    nkj = KeyJar().load(res)
+    assert set(nkj.owners()) == {'Alice', 'Bob', 'C'}
+    assert nkj.get_signing_key('rsa', 'Alice', kid="abc")
+    assert nkj.get_signing_key('rsa', 'C', kid='MnC_VZcATfM5pOYiJHMba9goEKY')