Updated tests.

Added tests for Context class.

Author: rohe

tests/test_04_message.py

@@ -1,25 +1,19 @@
+import json
 from urllib.parse import parse_qs
 from urllib.parse import urlparse
 
-import json
 import pytest
 from cryptojwt.exception import HeaderError
-
 from cryptojwt.jwk.hmac import SYMKey
 from cryptojwt.jwk.rsa import new_rsa_key
 from cryptojwt.jws.exception import NoSuitableSigningKeys
-
 from cryptojwt.key_bundle import KeyBundle
-from cryptojwt.key_jar import build_keyjar
 from cryptojwt.key_jar import KeyJar
+from cryptojwt.key_jar import build_keyjar
 
 from oidcmsg.exception import DecodeError
 from oidcmsg.exception import MessageException
 from oidcmsg.exception import OidcMsgError
-from oidcmsg.exception import WrongEncryptionAlgorithm
-
-from oidcmsg.message import json_deserializer, msg_ser
-from oidcmsg.message import json_serializer
 from oidcmsg.message import OPTIONAL_LIST_OF_MESSAGES
 from oidcmsg.message import OPTIONAL_LIST_OF_STRINGS
 from oidcmsg.message import OPTIONAL_MESSAGE
@@ -28,8 +22,10 @@
 from oidcmsg.message import SINGLE_OPTIONAL_JSON
 from oidcmsg.message import SINGLE_OPTIONAL_STRING
 from oidcmsg.message import SINGLE_REQUIRED_STRING
+from oidcmsg.message import json_deserializer
+from oidcmsg.message import json_serializer
+from oidcmsg.message import msg_ser
 from oidcmsg.message import sp_sep_list_deserializer
-
 from oidcmsg.oauth2 import Message
 
 __author__ = 'Roland Hedberg'
@@ -41,25 +37,25 @@
     {"type": "RSA", "use": ["enc"]},
     {"type": "EC", "crv": "P-256", "use": ["sig"]},
     {"type": "EC", "crv": "P-256", "use": ["enc"]},
-    ]
+]
 
 keym = [
     {"type": "RSA", "use": ["sig"]},
     {"type": "RSA", "use": ["sig"]},
     {"type": "RSA", "use": ["sig"]},
-    ]
+]
 
 KEYJAR = build_keyjar(keys)
 
 IKEYJAR = build_keyjar(keys)
-IKEYJAR.issuer_keys['issuer'] = IKEYJAR.issuer_keys['']
-del IKEYJAR.issuer_keys['']
+IKEYJAR.import_jwks(IKEYJAR.export_jwks(private=True), 'issuer')
+del IKEYJAR['']
 
 KEYJARS = {}
 for iss in ['A', 'B', 'C']:
     _kj = build_keyjar(keym)
-    _kj.issuer_keys[iss] = _kj.issuer_keys['']
-    del _kj.issuer_keys['']
+    _kj.import_jwks(_kj.export_jwks(private=True), iss)
+    del _kj['']
     KEYJARS[iss] = _kj
 
 
@@ -121,7 +117,7 @@ class DummyMessage(Message):
         "opt_str_list": OPTIONAL_LIST_OF_STRINGS,
         "req_str_list": REQUIRED_LIST_OF_STRINGS,
         "opt_json": SINGLE_OPTIONAL_JSON
-        }
+    }
 
 
 class TestMessage(object):
@@ -296,7 +292,7 @@ def test_int_instead_of_string(self):
 @pytest.mark.parametrize("keytype,alg", [
     ('RSA', 'RS256'),
     ('EC', 'ES256')
-    ])
+])
 def test_to_jwt(keytype, alg):
     msg = Message(a='foo', b='bar', c='tjoho')
     _jwt = msg.to_jwt(KEYJAR.get_signing_key(keytype, ''), alg)
@@ -307,7 +303,7 @@ def test_to_jwt(keytype, alg):
 @pytest.mark.parametrize("keytype,alg,enc", [
     ('RSA', 'RSA1_5', 'A128CBC-HS256'),
     ('EC', 'ECDH-ES', 'A128GCM'),
-    ])
+])
 def test_to_jwe(keytype, alg, enc):
     msg = Message(a='foo', b='bar', c='tjoho')
     _jwe = msg.to_jwe(KEYJAR.get_encrypt_key(keytype, ''), alg=alg, enc=enc)
@@ -334,7 +330,7 @@ class MsgMessage(Message):
         c_param = {
             "msg": OPTIONAL_MESSAGE,
             "opt_str": SINGLE_OPTIONAL_STRING,
-            }
+        }
 
     _dict = {
         "req_str": "Fair", "req_str_list": ["spike", "lee"],
@@ -359,7 +355,7 @@ class MsgMessage(Message):
         c_param = {
             "msgs": OPTIONAL_LIST_OF_MESSAGES,
             "opt_str": SINGLE_OPTIONAL_STRING,
-            }
+        }
 
     _dict = {
         "req_str": "Fair", "req_str_list": ["spike", "lee"],
@@ -385,7 +381,7 @@ class MsgMessage(Message):
         c_param = {
             "msgs": OPTIONAL_LIST_OF_MESSAGES,
             "opt_str": SINGLE_OPTIONAL_STRING,
-            }
+        }
 
     _dict = {
         "req_str": "Fair", "req_str_list": ["spike", "lee"],
@@ -409,7 +405,7 @@ class MsgMessage(Message):
         c_param = {
             "msgs": OPTIONAL_LIST_OF_MESSAGES,
             "opt_str": SINGLE_OPTIONAL_STRING,
-            }
+        }
 
     _dict = {
         "req_str": "Fair", "req_str_list": ["spike", "lee"],
@@ -446,7 +442,7 @@ def test_json_type_error():
 @pytest.mark.parametrize("keytype,alg,enc", [
     ('RSA', 'RSA1_5', 'A128CBC-HS256'),
     ('EC', 'ECDH-ES', 'A128GCM'),
-    ])
+])
 def test_to_jwe(keytype, alg, enc):
     msg = Message(a='foo', b='bar', c='tjoho')
     _jwe = msg.to_jwe(KEYJAR.get_encrypt_key(keytype, ''), alg=alg,
@@ -469,8 +465,7 @@ def test_no_suitable_keys():
     keytype = 'RSA'
     alg = 'RS256'
     msg = Message(a='foo', b='bar', c='tjoho')
-    _jwt = msg.to_jwt(NEW_KEYJAR.get_signing_key(keytype, '', kid=NEW_KID),
-                      alg)
+    _jwt = msg.to_jwt(NEW_KEYJAR.get_signing_key(keytype, '', kid=NEW_KID), alg)
     with pytest.raises(NoSuitableSigningKeys):
         Message().from_jwt(_jwt, KEYJAR)
 
@@ -495,9 +490,9 @@ def test_weed():
 def test_msg_ser():
     assert msg_ser('a.b.c', 'dict') == 'a.b.c'
     with pytest.raises(MessageException):
-        msg_ser([1,2], 'dict')
+        msg_ser([1, 2], 'dict')
     with pytest.raises(OidcMsgError):
-        msg_ser([1,2], 'list')
+        msg_ser([1, 2], 'list')
 
 
 def test_error_description():

tests/test_05_oauth2.py

@@ -42,14 +42,14 @@
 
 KEYJAR = build_keyjar(keys)
 IKEYJAR = build_keyjar(keys)
-IKEYJAR.issuer_keys['issuer'] = IKEYJAR.issuer_keys['']
-del IKEYJAR.issuer_keys['']
+IKEYJAR.import_jwks(IKEYJAR.export_jwks(private=True), 'issuer')
+del IKEYJAR['']
 
 KEYJARS = {}
 for iss in ['A', 'B', 'C']:
     _kj = build_keyjar(keym)
-    _kj.issuer_keys[iss] = _kj.issuer_keys['']
-    del _kj.issuer_keys['']
+    _kj.import_jwks(_kj.export_jwks(private=True) ,iss)
+    del _kj['']
     KEYJARS[iss] = _kj
 
 

tests/test_06_oidc.py

@@ -767,10 +767,9 @@ def test_ok_idtoken(self):
         idts = IdToken(**idval)
         keyjar = KeyJar()
         keyjar.add_symmetric('', "SomeTestPassword")
-        keyjar.add_symmetric('https://alpha.cloud.nds.rub.de',
-                             "SomeTestPassword")
-        _signed_jwt = idts.to_jwt(key=keyjar.get_signing_key('oct'),
-                                  algorithm="HS256", lifetime=300)
+        keyjar.add_symmetric('https://alpha.cloud.nds.rub.de', "SomeTestPassword")
+        _signed_jwt = idts.to_jwt(key=keyjar.get_signing_key('oct'), algorithm="HS256",
+                                  lifetime=300)
 
         _info = {
             "access_token": "accessTok", "id_token": _signed_jwt,

tests/test_07_session.py

@@ -49,11 +49,11 @@ def full_path(local_file):
 
 CLI_KEY = init_key_jar(public_path=full_path('pub_client.jwks'),
                        private_path=full_path('priv_client.jwks'),
-                       key_defs=KEYDEF, owner=CLIENT_ID)
+                       key_defs=KEYDEF, issuer_id=CLIENT_ID)
 
 ISS_KEY = init_key_jar(public_path=full_path('pub_iss.jwks'),
                        private_path=full_path('priv_iss.jwks'),
-                       key_defs=KEYDEF, owner=ISS)
+                       key_defs=KEYDEF, issuer_id=ISS)
 
 ISS_KEY.import_jwks_as_json(open(full_path('pub_client.jwks')).read(), CLIENT_ID)
 CLI_KEY.import_jwks_as_json(open(full_path('pub_iss.jwks')).read(),ISS)
@@ -325,7 +325,7 @@ def test_back_channel_logout_request():
         }
     lt = LogoutToken(**val)
     signer = JWS(lt.to_json(), alg='ES256')
-    _jws = signer.sign_compact(keys=ISS_KEY.get_signing_key(owner=ISS))
+    _jws = signer.sign_compact(keys=ISS_KEY.get_signing_key(issuer_id=ISS))
 
     bclr = BackChannelLogoutRequest(logout_token=_jws)
 
@@ -336,8 +336,7 @@ def test_back_channel_logout_request():
 
     assert 'logout_token' in _request
 
-    _verified = _request.verify(keyjar=CLI_KEY, iss=ISS,
-                                                     aud=CLIENT_ID, skew=30)
+    _verified = _request.verify(keyjar=CLI_KEY, iss=ISS, aud=CLIENT_ID, skew=30)
 
     assert _verified
     assert set(_request.keys()) == {'logout_token', '__verified_logout_token'}

tests/test_12_context.py

@@ -0,0 +1,87 @@
+import copy
+import shutil
+
+import pytest
+
+from oidcmsg.context import OidcContext
+
+KEYDEF = [
+    {"type": "EC", "crv": "P-256", "use": ["sig"]},
+    {"type": "EC", "crv": "P-256", "use": ["enc"]}
+]
+
+JWKS = {
+    "keys": [
+        {
+            "n":
+                'zkpUgEgXICI54blf6iWiD2RbMDCOO1jV0VSff1MFFnujM4othfMsad7H1kRo50YM5S'
+                '_X9TdvrpdOfpz5aBaKFhT6Ziv0nhtcekq1eRl8mjBlvGKCE5XGk-0LFSDwvqgkJoFY'
+                'Inq7bu0a4JEzKs5AyJY75YlGh879k1Uu2Sv3ZZOunfV1O1Orta-NvS-aG_jN5cstVb'
+                'CGWE20H0vFVrJKNx0Zf-u-aA-syM4uX7wdWgQ-owoEMHge0GmGgzso2lwOYf_4znan'
+                'LwEuO3p5aabEaFoKNR4K6GjQcjBcYmDEE4CtfRU9AEmhcD1kleiTB9TjPWkgDmT9MX'
+                'sGxBHf3AKT5w',
+            "e": "AQAB", "kty": "RSA", "kid": "rsa1"
+        },
+        {
+            "k":
+                'YTEyZjBlMDgxMGI4YWU4Y2JjZDFiYTFlZTBjYzljNDU3YWM0ZWNiNzhmNmFlYTNkNTY0NzMzYjE',
+            "kty": "oct"
+        },
+    ]
+}
+
+
+def test_context():
+    c = OidcContext({})
+    assert c.keyjar is not None
+
+
+class TestContext(object):
+    @pytest.fixture(autouse=True)
+    def setup(self):
+        try:
+            shutil.rmtree('db')
+        except FileNotFoundError:
+            pass
+
+        self.conf = {
+            'issuer': 'https://example.com',
+            'db_conf': {
+                'abstract_storage_cls': 'abstorage.extension.LabeledAbstractStorage',
+                'keyjar': {
+                    'handler': 'abstorage.storages.abfile.AbstractFileSystem',
+                    'fdir': 'db/keyjar',
+                    'key_conv': 'abstorage.converter.QPKey',
+                    'value_conv': 'cryptojwt.serialize.item.KeyIssuer',
+                    'label': 'keyjar'
+                },
+                'default': {
+                    'handler': 'abstorage.storages.abfile.AbstractFileSystem',
+                    'fdir': 'db',
+                    'key_conv': 'abstorage.converter.QPKey',
+                    'value_conv': 'abstorage.converter.JSON'
+                }
+            }
+        }
+
+    def test_context_with_entity_id_no_keys(self):
+        c = OidcContext(self.conf, entity_id='https://example.com')
+        assert c.keyjar.owners() == []
+
+    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'}
+
+    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')
+        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