State Import/Export base class.

Used for OidcContext

Author: rohe

src/oidcmsg/context.py

@@ -4,10 +4,8 @@
 from cryptojwt import KeyJar
 from cryptojwt.key_jar import init_key_jar
 
+from oidcmsg.impexp import ImpExp
 from oidcmsg.message import Message
-from oidcmsg.storage.init import get_storage_conf
-from oidcmsg.storage.init import init_storage
-from oidcmsg.storage.init import storage_factory
 
 
 def add_issuer(conf, issuer):
@@ -22,44 +20,24 @@ def add_issuer(conf, issuer):
     return res
 
 
-class OidcContext:
+class OidcContext(ImpExp):
+    parameter = {"keyjar": KeyJar, "issuer": None}
+
     def __init__(self, config=None, keyjar=None, entity_id=''):
+        ImpExp.__init__(self)
         if config is None:
             config = {}
 
-        self.db_conf = config.get('db_conf')
-        if self.db_conf:
-            _iss = config.get('issuer')
-            if _iss:
-                self.db_conf = add_issuer(self.db_conf, _iss)
-
-            if self.db_conf.get('default'):
-                self.db = init_storage(self.db_conf)
-            else:
-                self.db = init_storage()
-        else:
-            self.db = init_storage()
-
-        self.keyjar = self._keyjar(keyjar, self.db_conf, config, entity_id=entity_id)
-
-    def add_boxes(self, boxes, db_conf):
-        for key, attr in boxes.items():
-            setattr(self, attr, init_storage(db_conf, key))
+        self.issuer = entity_id
+        self.keyjar = self._keyjar(keyjar, conf=config, entity_id=entity_id)
 
-    def _keyjar(self, keyjar=None, db_conf=None, conf=None, entity_id=''):
+    def _keyjar(self, keyjar=None, conf=None, entity_id=''):
         if keyjar is None:
-            _storage = None
-            if db_conf:
-                _cnf = get_storage_conf(db_conf, 'keyjar')
-                if _cnf:
-                    _storage = storage_factory(_cnf)
-
             if 'keys' in conf:
                 args = {k: v for k, v in conf["keys"].items() if k != "uri_path"}
-                args.update({'storage': _storage})
                 _keyjar = init_key_jar(**args)
             else:
-                _keyjar = KeyJar(storage=_storage)
+                _keyjar = KeyJar()
                 if 'jwks' in conf:
                     _keyjar.import_jwks(conf['jwks'], '')
 
@@ -74,15 +52,3 @@ def _keyjar(self, keyjar=None, db_conf=None, conf=None, entity_id=''):
             return _keyjar
         else:
             return keyjar
-
-    def set(self, item, value):
-        if isinstance(value, Message):
-            self.db[item] = value.to_dict()
-        else:
-            self.db[item] = value
-
-    def get(self, item):
-        if item == 'seed':
-            return bytes(self.db[item], 'utf-8')
-        else:
-            return self.db[item]

src/oidcmsg/impexp.py

@@ -0,0 +1,102 @@
+import copy
+from typing import Optional
+from urllib.parse import quote_plus
+
+from cryptojwt.utils import importer
+from cryptojwt.utils import qualified_name
+
+from oidcmsg.message import Message
+
+
+def add_issuer(conf, issuer):
+    res = {}
+    for key, val in conf.items():
+        if key == 'abstract_storage_cls':
+            res[key] = val
+        else:
+            _val = copy.copy(val)
+            _val['issuer'] = quote_plus(issuer)
+            res[key] = _val
+    return res
+
+
+class ImpExp:
+    parameter = {}
+
+    def __init__(self):
+        pass
+
+    def _dump(self, cls, item, cutoff: Optional[list] = None):
+        if cls in [None, "", [], {}]:
+            val = item
+        elif isinstance(item, Message):
+            val = {qualified_name(item.__class__): item.to_dict()}
+        elif cls == object:
+            val = qualified_name(item)
+        elif isinstance(cls, list):
+            val = [self._dump(cls[0], v, cutoff) for v in item]
+        else:
+            val = item.dump(cutoff=cutoff)
+
+        return val
+
+    def dump(self, cutoff: Optional[list] = None) -> dict:
+        _cutoff = cutoff or []
+        info = {}
+        for attr, cls in self.parameter.items():
+            if attr in _cutoff:
+                continue
+
+            item = getattr(self, attr, None)
+            if item is None:
+                continue
+
+            info[attr] = self._dump(cls, item, cutoff)
+
+        return info
+
+    def _local_adjustments(self):
+        pass
+
+    def _load(self, cls, item):
+        if cls in [None, "", [], {}]:
+            val = item
+        elif cls == object:
+            val = importer(item)
+        elif isinstance(cls, list):
+            val = [cls[0]().load(v) for v in item]
+        elif issubclass(cls, Message):
+            val = cls().from_dict(item)
+        else:
+            val = cls().load(item)
+
+        return val
+
+    def load(self, item: dict):
+        for attr, cls in self.parameter.items():
+            if attr not in item:
+                continue
+
+            setattr(self, attr, self._load(cls, item[attr]))
+
+        self._local_adjustments()
+        return self
+
+    def flush(self):
+        """
+        Reset the content of the instance to its pristine state
+
+        :return: A reference to the instance itself
+        """
+        for attr, cls in self.parameter.items():
+            if cls is None:
+                setattr(self, attr, None)
+            elif cls == "":
+                setattr(self, attr, "")
+            elif cls == []:
+                setattr(self, attr, [])
+            elif cls == {}:
+                setattr(self, attr, {})
+            else:
+                setattr(self, attr, None)
+        return self

src/oidcmsg/storage/converter.py

@@ -12,8 +12,8 @@ def deserialize(self, str):
 
 
 class JSON:
-    def serialize(self, str):
-        return json.dumps(str)
+    def serialize(self, item):
+        return json.dumps(item)
 
     def deserialize(self, str):
         return json.loads(str)

tests/data/keys/cert.key

@@ -31,56 +31,62 @@
 }
 
 
-def test_context():
+def test_dump_load():
     c = OidcContext({})
     assert c.keyjar is not None
+    mem = c.dump()
+    c2 = OidcContext().load(mem)
+    assert c2.keyjar is not None
 
 
-class TestContext(object):
+class TestDumpLoad(object):
     @pytest.fixture(autouse=True)
     def setup(self):
-        try:
-            shutil.rmtree('db')
-        except FileNotFoundError:
-            pass
-
         self.conf = {
-            'issuer': 'https://example.com',
-            'db_conf': {
-                'keyjar': {
-                    'handler': 'oidcmsg.storage.abfile.LabeledAbstractFileSystem',
-                    'fdir': 'db/keyjar',
-                    'key_conv': 'oidcmsg.storage.converter.QPKey',
-                    'value_conv': 'cryptojwt.serialize.item.KeyIssuer',
-                    'label': 'foo'
-                },
-                'default': {
-                    'handler': 'oidcmsg.storage.abfile.AbstractFileSystem',
-                    'fdir': 'db',
-                    'key_conv': 'oidcmsg.storage.converter.QPKey',
-                    'value_conv': 'oidcmsg.storage.converter.JSON'
-                }
-            }
+            'issuer': 'https://example.com'
         }
 
     def test_context_with_entity_id_no_keys(self):
         c = OidcContext(self.conf, entity_id='https://example.com')
-        assert c.keyjar.owners() == []
+        mem = c.dump()
+        c2 = OidcContext().load(mem)
+        assert c2.keyjar.owners() == []
+        assert c2.issuer == 'https://example.com'
 
     def test_context_with_entity_id_and_keys(self):
         conf = copy.deepcopy(self.conf)
         conf['keys'] = {'key_defs': KEYDEF}
-
         c = OidcContext(conf, entity_id='https://example.com')
-        assert set(c.keyjar.owners()) == {'', 'https://example.com'}
+
+        mem = c.dump()
+        c2 = OidcContext().load(mem)
+        assert set(c2.keyjar.owners()) == {'', 'https://example.com'}
 
     def test_context_with_entity_id_and_jwks(self):
         conf = copy.deepcopy(self.conf)
         conf['jwks'] = JWKS
+        c = OidcContext(conf, entity_id='https://example.com')
+
+        mem = c.dump()
+        c2 = OidcContext().load(mem)
+
+        assert set(c2.keyjar.owners()) == {'', 'https://example.com'}
+        assert len(c2.keyjar.get('sig', 'RSA')) == 1
+        assert len(c2.keyjar.get('sig', 'RSA', issuer_id='https://example.com')) == 1
+        assert len(c2.keyjar.get('sig', 'oct')) == 1
+        assert len(c2.keyjar.get('sig', 'oct', issuer_id='https://example.com')) == 1
+
+    def test_context_restore(self):
+        conf = copy.deepcopy(self.conf)
+        conf['keys'] = {'key_defs': KEYDEF}
 
         c = OidcContext(conf, entity_id='https://example.com')
-        assert set(c.keyjar.owners()) == {'', 'https://example.com'}
-        assert len(c.keyjar.get('sig', 'RSA')) == 1
-        assert len(c.keyjar.get('sig', 'RSA', issuer_id='https://example.com')) == 1
-        assert len(c.keyjar.get('sig', 'oct')) == 1
-        assert len(c.keyjar.get('sig', 'oct', issuer_id='https://example.com')) == 1
+        mem = c.dump()
+        c2 = OidcContext().load(mem)
+
+        assert set(c2.keyjar.owners()) == {'', 'https://example.com'}
+        assert len(c2.keyjar.get('sig', 'EC')) == 1
+        assert len(c2.keyjar.get('enc', 'EC')) == 1
+        assert len(c.keyjar.get('sig', 'RSA')) == 0
+        assert len(c.keyjar.get('sig', 'oct')) == 0
+