Added more tests.

As a consequence some code changes was made.

Author: rohe

chrp/rp.py

@@ -1,14 +1,11 @@
 #!/usr/bin/env python3
+import cherrypy
 import importlib
-import json
 import logging
 import os
 import sys
 
-import cherrypy
-
-from oidcmsg.key_jar import build_keyjar, init_key_jar
-from oidcmsg.key_jar import KeyJar
+from cryptojwt.key_jar import init_key_jar
 
 from oidcrp import RPHandler
 
@@ -22,6 +19,7 @@
 logger.addHandler(hdlr)
 logger.setLevel(logging.DEBUG)
 
+
 SIGKEY_NAME = 'sigkey.jwks'
 
 

src/oidcrp/__init__.py

@@ -70,9 +70,7 @@ def load_registration_response(client):
 
     :param client: A :py:class:`oidcservice.oidc.Client` instance
     """
-    try:
-        _client_reg = client.service_context.config['registration_response']
-    except KeyError:
+    if not client.service_context.client_id:
         try:
             response = client.do_request('registration')
         except KeyError:
@@ -83,8 +81,6 @@ def load_registration_response(client):
         else:
             if 'error' in response:
                 raise OidcServiceError(response.to_json())
-    else:
-        client.service_context.registration_info = _client_reg
 
 
 def dynamic_provider_info_discovery(client):
@@ -114,7 +110,7 @@ class RPHandler(object):
     def __init__(self, base_url='', hash_seed="", keyjar=None, verify_ssl=True,
                  services=None, service_factory=None, client_configs=None,
                  client_authn_factory=None, client_cls=None,
-                 state_db=None, **kwargs):
+                 state_db=None, http_lib=None, **kwargs):
         self.base_url = base_url
         self.hash_seed = as_bytes(hash_seed)
         self.verify_ssl = verify_ssl
@@ -143,6 +139,7 @@ def __init__(self, base_url='', hash_seed="", keyjar=None, verify_ssl=True,
         # keep track on which RP instance that serves with OP
         self.issuer2rp = {}
         self.hash2issuer = {}
+        self.httplib = http_lib
 
     def supports_webfinger(self):
         """
@@ -209,7 +206,9 @@ def init_client(self, issuer):
                 state_db=self.state_db,
                 client_authn_factory=self.client_authn_factory,
                 verify_ssl=self.verify_ssl, services=_services,
-                service_factory=self.service_factory, config=_cnf)
+                service_factory=self.service_factory, config=_cnf,
+                httplib=self.httplib
+            )
         except Exception as err:
             logger.error('Failed initiating client: {}'.format(err))
             message = traceback.format_exception(*sys.exc_info())
@@ -413,7 +412,7 @@ def begin(self, issuer_id='', user_id=''):
 
         try:
             res = self.init_authorization(client)
-        except Exception as err:
+        except Exception:
             message = traceback.format_exception(*sys.exc_info())
             logger.error(message)
             raise

src/oidcrp/http.py

@@ -1,13 +1,11 @@
 import copy
 import logging
+import requests
+
 from http.cookiejar import FileCookieJar
 from http.cookies import CookieError
 from http.cookies import SimpleCookie
 
-import requests
-
-from cryptojwt.key_jar import KeyJar
-
 from oidcservice import sanitize
 from oidcservice.exception import NonFatalException
 
@@ -19,23 +17,19 @@
 
 
 class HTTPLib(object):
-    def __init__(self, ca_certs=None, verify_ssl=True, keyjar=None,
-                 client_cert=None):
+    def __init__(self, ca_certs=None, verify_ssl=True, client_cert=None):
         """
         A base class for OAuth2 clients and servers
 
         :param ca_certs: the path to a CA_BUNDLE file or directory with
             certificates of trusted CAs
         :param verify_ssl: If True then the server SSL certificate is not
             verfied
-        :param keyjar: A place to keep keys for signing/encrypting messages
         :param client_cert: local cert to use as client side certificate, as a
             single file (containing the private key and the certificate) or as
             a tuple of both file's path
         """
 
-        self.keyjar = keyjar or KeyJar(verify_ssl=verify_ssl)
-
         self.request_args = {"allow_redirects": False}
 
         self.cookiejar = FileCookieJar()
@@ -143,11 +137,3 @@ def send(self, url, method="GET", **kwargs):
         :return: Request response
         """
         return self(url, method, **kwargs)
-
-    def load_cookies_from_file(self, filename, ignore_discard=False,
-                               ignore_expires=False):
-        self.cookiejar.load(filename, ignore_discard, ignore_expires)
-
-    def save_cookies_to_file(self, filename, ignore_discard=False,
-                             ignore_expires=False):
-        self.cookiejar.save(filename, ignore_discard, ignore_expires)

src/oidcrp/oauth2/__init__.py

@@ -3,6 +3,7 @@
 import cherrypy
 
 from cryptojwt.key_jar import KeyJar
+from oidcmsg.exception import FormatError
 
 from oidcservice.client_auth import factory as ca_factory
 from oidcservice.exception import OidcServiceError
@@ -65,8 +66,7 @@ def __init__(self, state_db, ca_certs=None, client_authn_factory=None,
         self.session_interface = StateInterface(state_db)
         self.http = httplib or HTTPLib(ca_certs=ca_certs,
                                        verify_ssl=verify_ssl,
-                                       client_cert=client_cert,
-                                       keyjar=keyjar)
+                                       client_cert=client_cert)
 
         if not keyjar:
             keyjar = KeyJar()
@@ -90,16 +90,6 @@ def __init__(self, state_db, ca_certs=None, client_authn_factory=None,
 
         self.verify_ssl = verify_ssl
 
-    def construct(self, request_type, request_args=None, extra_args=None,
-                  **kwargs):
-        try:
-            self.service[request_type]
-        except KeyError:
-            raise NotImplemented(request_type)
-
-        method = getattr(self, 'construct_{}_request'.format(request_type))
-        return method(self.service_context, request_args, extra_args, **kwargs)
-
     def do_request(self, request_type, response_body_type="", request_args=None,
                    **kwargs):
 
@@ -222,12 +212,12 @@ def parse_request_response(self, service, reqresp, response_body_type='',
 
             try:
                 err_resp = service.parse_response(reqresp.text, _deser_method)
-            except OidcServiceError:
+            except FormatError:
                 if _deser_method != response_body_type:
                     try:
                         err_resp = service.parse_response(reqresp.text,
                                                           response_body_type)
-                    except OidcServiceError:
+                    except (OidcServiceError, FormatError):
                         raise cherrypy.HTTPError("HTTP ERROR: %s [%s] on %s" % (
                             reqresp.text, reqresp.status_code, reqresp.url))
                 else:

src/oidcrp/util.py

@@ -122,25 +122,6 @@ def set_cookie(cookiejar, kaka):
             cookiejar.set_cookie(new_cookie)
 
 
-def get_response_body_type(response):
-    try:
-        _ctype = response.headers["content-type"]
-    except KeyError:
-        raise ValueError('Missing Content-type specification')
-
-    body_type = ''
-
-    if match_to_("application/json", _ctype) or match_to_(
-            'application/jrd+json', _ctype):
-        body_type = 'json'
-    elif match_to_("application/jwt", _ctype):
-        body_type = "jwt"
-    elif match_to_(URL_ENCODED, _ctype):
-        body_type = 'urlencoded'
-
-    return body_type
-
-
 def verify_header(reqresp, body_type):
     """
     
@@ -215,7 +196,7 @@ def get_deserialization_method(reqresp):
     try:
         _ctype = reqresp.headers["content-type"]
     except KeyError:
-        return 'txt'  # reasonable default ??
+        return 'urlencoded'  # reasonable default ??
 
     if match_to_("application/json", _ctype) or match_to_(
             'application/jrd+json', _ctype):
@@ -232,21 +213,21 @@ def get_deserialization_method(reqresp):
     return deser_method
 
 
-SORT_ORDER = {'RS': 0, 'ES': 1, 'HS': 2, 'PS': 3, 'no': 4}
-
-
-def sort_sign_alg(alg1, alg2):
-    if SORT_ORDER[alg1[0:2]] < SORT_ORDER[alg2[0:2]]:
-        return -1
-    elif SORT_ORDER[alg1[0:2]] > SORT_ORDER[alg2[0:2]]:
-        return 1
-    else:
-        if alg1 < alg2:
-            return -1
-        elif alg1 > alg2:
-            return 1
-        else:
-            return 0
+# SORT_ORDER = {'RS': 0, 'ES': 1, 'HS': 2, 'PS': 3, 'no': 4}
+#
+#
+# def sort_sign_alg(alg1, alg2):
+#     if SORT_ORDER[alg1[0:2]] < SORT_ORDER[alg2[0:2]]:
+#         return -1
+#     elif SORT_ORDER[alg1[0:2]] > SORT_ORDER[alg2[0:2]]:
+#         return 1
+#     else:
+#         if alg1 < alg2:
+#             return -1
+#         elif alg1 > alg2:
+#             return 1
+#         else:
+#             return 0
 
 
 def get_value_type(http_response, body_type):

tests/test_01_base.py

@@ -0,0 +1,35 @@
+from oidcrp import add_path
+from oidcrp import load_registration_response
+from oidcrp.oidc import RP
+
+
+class DB(object):
+    def __init__(self):
+        self.db = {}
+
+    def set(self, key, value):
+        self.db[key] = value
+
+    def get(self, item):
+        return self.db[item]
+
+
+def test_add_path():
+    assert add_path('https://example.com/', '/usr') == 'https://example.com/usr'
+    assert add_path('https://example.com/', 'usr') == 'https://example.com/usr'
+    assert add_path('https://example.com', '/usr') == 'https://example.com/usr'
+    assert add_path('https://example.com', 'usr') == 'https://example.com/usr'
+
+
+def test_load_registration_response():
+    conf = {
+        'redirect_uris': ['https://example.com/cli/authz_cb'],
+        'client_id': 'client_1',
+        'client_secret': 'abcdefghijklmnop',
+        'registration_response': {'issuer': 'https://example.com'}
+    }
+    client = RP(DB(), config=conf)
+
+    # test static
+    load_registration_response(client)
+    assert True

tests/test_03_util.py

@@ -6,7 +6,7 @@
 from urllib.parse import parse_qs
 from urllib.parse import urlparse
 
-
+from oidcrp.util import get_deserialization_method, URL_ENCODED
 from oidcservice.exception import WrongContentType
 
 from oidcrp import util
@@ -101,12 +101,13 @@ def test_match_to():
     assert not util.match_to_(str3, list_of_str)
 
 
-def test_verify_header():
-    class FakeResponse():
-        def __init__(self, header):
-            self.headers = {"content-type": header}
-            self.text = "TEST_RESPONSE"
+class FakeResponse():
+    def __init__(self, header):
+        self.headers = {"content-type": header}
+        self.text = "TEST_RESPONSE"
+
 
+def test_verify_header():
     json_header = "application/json"
     jwt_header = "application/jwt"
     default_header = util.DEFAULT_POST_CONTENT_TYPE
@@ -120,6 +121,15 @@ def __init__(self, header):
                               "urlencoded") == "urlencoded"
     assert util.verify_header(FakeResponse(plain_text_header),
                               "urlencoded") == "urlencoded"
+    assert util.verify_header(FakeResponse('text/html'), 'txt')
+    assert util.verify_header(FakeResponse('text/plain'), 'txt')
+
+    assert util.verify_header(FakeResponse(json_header), "") == "json"
+    assert util.verify_header(FakeResponse(jwt_header), "") == "jwt"
+    assert util.verify_header(FakeResponse(jwt_header), "") == "jwt"
+    assert util.verify_header(FakeResponse(default_header), "") == "urlencoded"
+    assert util.verify_header(FakeResponse(plain_text_header), "") == "txt"
+    assert util.verify_header(FakeResponse('text/html'), '') == 'txt'
 
     with pytest.raises(WrongContentType):
         util.verify_header(FakeResponse(json_header), "urlencoded")
@@ -130,3 +140,38 @@ def __init__(self, header):
 
     with pytest.raises(ValueError):
         util.verify_header(FakeResponse(json_header), "undefined")
+
+
+def test_get_deserialization_method_json():
+    resp = FakeResponse("application/json")
+    assert get_deserialization_method(resp) == 'json'
+
+    resp = FakeResponse("application/json; charset=utf-8")
+    assert get_deserialization_method(resp) == 'json'
+
+    resp.headers["content-type"] = "application/jrd+json"
+    assert get_deserialization_method(resp) == 'json'
+
+
+def test_get_deserialization_method_jwt():
+    resp = FakeResponse("application/jwt")
+    assert get_deserialization_method(resp) == 'jwt'
+
+
+def test_get_deserialization_method_urlencoded():
+    resp = FakeResponse(URL_ENCODED)
+    assert get_deserialization_method(resp) == 'urlencoded'
+
+
+def test_get_deserialization_method_text():
+    resp = FakeResponse('text/html')
+    assert get_deserialization_method(resp) == ''
+
+    resp = FakeResponse('text/plain')
+    assert get_deserialization_method(resp) == ''
+
+
+def test_verify_no_content_type():
+    resp = FakeResponse('text/html')
+    del resp.headers['content-type']
+    assert util.verify_header(resp, 'txt') == 'txt'