refactoring

Author: rohe

src/idpyoidc/client/entity_metadata.py

@@ -1,9 +1,12 @@
 from typing import Optional
 
+from idpyoidc.impexp import ImpExp
 
-class EntityMetadata(object):
 
+class EntityMetadata(ImpExp):
+    parameter = {"metadata": {}}
     def __init__(self, metadata: Optional[dict] = None):
+        ImpExp.__init__(self)
         if metadata is None:
             self.metadata = {}
         else:
@@ -20,20 +23,14 @@ def __setitem__(self, key, value):
         # Assumes not multiple entity types
         self.metadata[key] = value
 
-    def entity_types(self):
-        return list(self.metadata.keys())
-
-    def entity_type(self, etype):
-        return self.metadata[etype]
-
     def items(self):
         return self.metadata.items()
 
-    def get_entity_type_claim(self, entity_type, claim):
-        return self.metadata[entity_type][claim]
-
     def __contains__(self, item):
         return item in self.metadata
 
     def get(self, item, default=None):
         return self.metadata.get(item, default)
+
+    def to_dict(self):
+        return self.metadata

src/idpyoidc/client/oauth2/add_on/jar.py

@@ -5,6 +5,7 @@
 from idpyoidc import metadata
 from idpyoidc.client.oidc.utils import construct_request_uri
 from idpyoidc.client.oidc.utils import request_object_encryption
+from idpyoidc.client.request_object import construct_request_parameter
 from idpyoidc.message.oidc import make_openid_request
 from idpyoidc.time_util import utc_time_sans_frac
 
@@ -35,93 +36,6 @@ def store_request_on_file(service, req, **kwargs):
     return _webname
 
 
-def get_request_object_signing_alg(service, **kwargs):
-    alg = ""
-    for arg in ["request_object_signing_alg", "algorithm"]:
-        try:  # Trumps everything
-            alg = kwargs[arg]
-        except KeyError:
-            pass
-        else:
-            break
-
-    if not alg:
-        _context = service.upstream_get("context")
-        alg = _context.add_on["jar"].get("request_object_signing_alg")
-        if alg is None:
-            alg = "RS256"
-    return alg
-
-
-def construct_request_parameter(service, req, audience=None, **kwargs):
-    """Construct a request parameter"""
-    alg = get_request_object_signing_alg(service, **kwargs)
-    kwargs["request_object_signing_alg"] = alg
-
-    _context = service.upstream_get("context")
-    if "keys" not in kwargs and alg and alg != "none":
-        kwargs["keys"] = service.upstream_get("attribute", "keyjar")
-
-    if alg == "none":
-        kwargs["keys"] = []
-
-    # This is the issuer of the JWT, that is me !
-    _issuer = kwargs.get("issuer")
-    if _issuer is None:
-        kwargs["issuer"] = _context.get_client_id()
-
-    if kwargs.get("recv") is None:
-        try:
-            kwargs["recv"] = _context.provider_info["issuer"]
-        except KeyError:
-            kwargs["recv"] = _context.issuer
-
-    try:
-        del kwargs["service"]
-    except KeyError:
-        pass
-
-    _jar_conf = _context.add_on["jar"]
-    expires_in = _jar_conf.get("expires_in", DEFAULT_EXPIRES_IN)
-    if expires_in:
-        req["exp"] = utc_time_sans_frac() + int(expires_in)
-
-    if _jar_conf.get("with_jti", False):
-        kwargs["with_jti"] = True
-
-    _enc_enc = _jar_conf.get("request_object_encryption_enc", "")
-    if _enc_enc:
-        kwargs["request_object_encryption_enc"] = _enc_enc
-        kwargs["request_object_encryption_alg"] = _jar_conf.get("request_object_encryption_alg")
-
-    # Filter out only the arguments I want
-    _mor_args = {
-        k: kwargs[k]
-        for k in [
-            "keys",
-            "issuer",
-            "request_object_signing_alg",
-            "recv",
-            "with_jti",
-            "lifetime",
-        ]
-        if k in kwargs
-    }
-
-    if audience:
-        _mor_args["aud"] = audience
-
-    _req_jwt = make_openid_request(req, **_mor_args)
-
-    if "target" not in kwargs:
-        kwargs["target"] = _context.provider_info.get("issuer", _context.issuer)
-
-    # Should the request be encrypted
-    _req_jwte = request_object_encryption(
-        _req_jwt, _context, service.upstream_get("attribute", "keyjar"), **kwargs
-    )
-    return _req_jwte
-
 
 def jar_post_construct(request_args, service, **kwargs):
     """

src/idpyoidc/client/oidc/__init__.py

@@ -81,18 +81,18 @@ class RP(oauth2.Client):
     client_type = "oidc"
 
     def __init__(
-        self,
-        keyjar: Optional[KeyJar] = None,
-        config: Optional[Union[dict, Configuration]] = None,
-        services: Optional[dict] = None,
-        httpc: Optional[Callable] = None,
-        httpc_params: Optional[dict] = None,
-        upstream_get: Optional[Callable] = None,
-        key_conf: Optional[dict] = None,
-        entity_id: Optional[str] = "",
-        verify_ssl: Optional[bool] = True,
-        jwks_uri: Optional[str] = "",
-        **kwargs
+            self,
+            keyjar: Optional[KeyJar] = None,
+            config: Optional[Union[dict, Configuration]] = None,
+            services: Optional[dict] = None,
+            httpc: Optional[Callable] = None,
+            httpc_params: Optional[dict] = None,
+            upstream_get: Optional[Callable] = None,
+            key_conf: Optional[dict] = None,
+            entity_id: Optional[str] = "",
+            verify_ssl: Optional[bool] = True,
+            jwks_uri: Optional[str] = "",
+            **kwargs
     ):
         if services:
             _srvs = services

src/idpyoidc/client/oidc/authorization.py

@@ -8,17 +8,15 @@
 from idpyoidc.client.oauth2.utils import pre_construct_pick_redirect_uri
 from idpyoidc.client.oidc import IDT2REG
 from idpyoidc.client.oidc.utils import construct_request_uri
-from idpyoidc.client.oidc.utils import request_object_encryption
+from idpyoidc.client.request_object import construct_request_parameter
 from idpyoidc.client.service_context import ServiceContext
 from idpyoidc.client.util import implicit_response_types
 from idpyoidc.exception import MissingRequiredAttribute
 from idpyoidc.message import Message
 from idpyoidc.message import oauth2
 from idpyoidc.message import oidc
-from idpyoidc.message.oidc import make_openid_request
 from idpyoidc.message.oidc import verified_claim_name
 from idpyoidc.time_util import time_sans_frac
-from idpyoidc.time_util import utc_time_sans_frac
 from idpyoidc.util import rndstr
 
 __author__ = "Roland Hedberg"
@@ -212,63 +210,6 @@ def store_request_on_file(self, req, **kwargs):
         fid.close()
         return _webname
 
-    def construct_request_parameter(
-        self, req, request_param, audience=None, expires_in=0, **kwargs
-    ):
-        """Construct a request parameter"""
-        alg = self.get_request_object_signing_alg(**kwargs)
-        kwargs["request_object_signing_alg"] = alg
-
-        _context = self.upstream_get("context")
-        if "keys" not in kwargs and alg and alg != "none":
-            kwargs["keys"] = self.upstream_get("attribute", "keyjar")
-
-        if alg == "none":
-            kwargs["keys"] = []
-
-        # This is the issuer of the JWT, that is me !
-        _issuer = kwargs.get("issuer")
-        if _issuer is None:
-            kwargs["issuer"] = _context.get_client_id()
-
-        if kwargs.get("recv") is None:
-            try:
-                kwargs["recv"] = _context.provider_info["issuer"]
-            except KeyError:
-                kwargs["recv"] = _context.issuer
-
-        try:
-            del kwargs["service"]
-        except KeyError:
-            pass
-
-        if expires_in:
-            req["exp"] = utc_time_sans_frac() + int(expires_in)
-
-        _mor_args = {
-            k: kwargs[k]
-            for k in [
-                "keys",
-                "issuer",
-                "request_object_signing_alg",
-                "recv",
-                "with_jti",
-                "lifetime",
-            ]
-            if k in kwargs
-        }
-
-        _req_jwt = make_openid_request(req, **_mor_args)
-
-        if "target" not in kwargs:
-            kwargs["target"] = _context.provider_info.get("issuer", _context.issuer)
-
-        # Should the request be encrypted
-        _req_jwte = request_object_encryption(
-            _req_jwt, _context, self.upstream_get("attribute", "keyjar"), **kwargs
-        )
-        return _req_jwte
-
     def oidc_post_construct(self, req, **kwargs):
         """
         Modify the request arguments.
@@ -303,10 +244,21 @@ def oidc_post_construct(self, req, **kwargs):
         if _request_param == "request_uri":
             kwargs["base_path"] = _context.get("base_url") + "/" + "requests"
             kwargs["local_dir"] = _context.get_usage("requests_dir", "./requests")
-            _req = self.construct_request_parameter(req, _request_param, **kwargs)
+            service = kwargs.get("service")
+            if service:
+                del kwargs["service"]
+            else:
+                service = self
+
+            _req = construct_request_parameter(service, req, _request_param, **kwargs)
             req["request_uri"] = self.store_request_on_file(_req, **kwargs)
         elif _request_param == "request":
-            _req = self.construct_request_parameter(req, _request_param, **kwargs)
+            service = kwargs.get("service")
+            if service:
+                del kwargs["service"]
+            else:
+                service = self
+            _req = construct_request_parameter(service, req, _request_param, **kwargs)
             req["request"] = _req
 
         if _req:
@@ -319,7 +271,8 @@ def oidc_post_construct(self, req, **kwargs):
         return req
 
     def gather_verify_arguments(
-        self, response: Optional[Union[dict, Message]] = None, behaviour_args: Optional[dict] = None
+            self, response: Optional[Union[dict, Message]] = None,
+            behaviour_args: Optional[dict] = None
     ):
         """
         Need to add some information before running verify()
@@ -379,12 +332,12 @@ def _do_type(self, context, typ, response_types):
         return ""
 
     def construct_uris(
-        self,
-        base_url: str,
-        hex: bytes,
-        context: ServiceContext,
-        targets: Optional[List[str]] = None,
-        response_types: Optional[List[str]] = None,
+            self,
+            base_url: str,
+            hex: bytes,
+            context: ServiceContext,
+            targets: Optional[List[str]] = None,
+            response_types: Optional[List[str]] = None,
     ):
         _callback_uris = context.get_preference("callback_uris", {})