Typing

Author: rohe

src/oidcrp/client_auth.py

@@ -12,8 +12,8 @@
 from oidcmsg.oauth2 import SINGLE_OPTIONAL_STRING
 from oidcmsg.oidc import AuthnToken
 from oidcmsg.time_util import utc_time_sans_frac
+from oidcmsg.util import rndstr
 
-from oidcrp.util import rndstr
 from oidcrp.util import sanitize
 from .defaults import DEF_SIGN_ALG
 from .defaults import JWT_BEARER

src/oidcrp/oauth2/__init__.py

@@ -1,15 +1,19 @@
 from json import JSONDecodeError
 import logging
+from typing import Optional
 
 from oidcmsg.exception import FormatError
+from oidcmsg.message import Message
+from oidcmsg.oauth2 import is_error_message
 
-from oidcrp.configure import URIS
 from oidcrp.entity import Entity
+from oidcrp.exception import ConfigurationError
 from oidcrp.exception import OidcServiceError
 from oidcrp.exception import ParseError
 from oidcrp.http import HTTPLib
 from oidcrp.service import REQUEST_INFO
 from oidcrp.service import SUCCESSFUL
+from oidcrp.service import Service
 from oidcrp.util import do_add_ons
 from oidcrp.util import get_deserialization_method
 
@@ -73,7 +77,11 @@ def __init__(self, client_authn_factory=None, keyjar=None, verify_ssl=True, conf
         # just ignore verify_ssl until it goes away
         self.verify_ssl = self.httpc_params.get("verify", True)
 
-    def do_request(self, request_type, response_body_type="", request_args=None, **kwargs):
+    def do_request(self,
+                   request_type: str,
+                   response_body_type: Optional[str] = "",
+                   request_args: Optional[dict] = None,
+                   behaviour_args: Optional[dict] = None, **kwargs):
         _srv = self._service[request_type]
 
         _info = _srv.get_request_parameters(request_args=request_args, **kwargs)
@@ -94,8 +102,13 @@ def set_client_id(self, client_id):
         self.client_id = client_id
         self._service_context.set('client_id', client_id)
 
-    def get_response(self, service, url, method="GET", body=None, response_body_type="",
-                     headers=None, **kwargs):
+    def get_response(self,
+                     service: Service,
+                     url: str,
+                     method: Optional[str] = "GET",
+                     body: Optional[dict] = None,
+                     response_body_type: Optional[str] = "",
+                     headers: Optional[dict] = None, **kwargs):
         """
 
         :param url:
@@ -130,8 +143,13 @@ def get_response(self, service, url, method="GET", body=None, response_body_type
         return self.parse_request_response(service, resp,
                                            response_body_type, **kwargs)
 
-    def service_request(self, service, url, method="GET", body=None,
-                        response_body_type="", headers=None, **kwargs):
+    def service_request(self,
+                        service: Service,
+                        url: str,
+                        method: Optional[str] = "GET",
+                        body: Optional[dict] = None,
+                        response_body_type: Optional[str] = "",
+                        headers: Optional[dict] = None, **kwargs) -> Message:
         """
         The method that sends the request and handles the response returned.
         This assumes that the response arrives in the HTTP response.
@@ -250,3 +268,27 @@ def parse_request_response(self, service, reqresp, response_body_type='',
                                                           reqresp.text))
             raise OidcServiceError("HTTP ERROR: %s [%s] on %s" % (
                 reqresp.text, reqresp.status_code, reqresp.url))
+
+
+def dynamic_provider_info_discovery(client: Client, behaviour_args: Optional[dict]=None):
+    """
+    This is about performing dynamic Provider Info discovery
+
+    :param behaviour_args:
+    :param client: A :py:class:`oidcrp.oidc.Client` instance
+    """
+    try:
+        client.get_service('provider_info')
+    except KeyError:
+        raise ConfigurationError(
+            'Can not do dynamic provider info discovery')
+    else:
+        _context = client.client_get("service_context")
+        try:
+            _context.set('issuer', _context.config['srv_discovery_url'])
+        except KeyError:
+            pass
+
+        response = client.do_request('provider_info', behaviour_args=behaviour_args)
+        if is_error_message(response):
+            raise OidcServiceError(response['error'])

src/oidcrp/rp_handler.py

@@ -28,9 +28,9 @@
 from .defaults import DEFAULT_RP_KEY_DEFS
 from .exception import OidcServiceError
 from .oauth2 import Client
+from .oauth2 import dynamic_provider_info_discovery
 from .oauth2.utils import pick_redirect_uri
 from .util import add_path
-from .util import dynamic_provider_info_discovery
 from .util import load_registration_response
 from .util import rndstr
 
@@ -185,11 +185,15 @@ def init_client(self, issuer):
         _context.jwks_uri = self.jwks_uri
         return client
 
-    def do_provider_info(self, client=None, state='', behaviour_args=None):
+    def do_provider_info(self,
+                         client: Optional[Client]=None,
+                         state: Optional[str]='',
+                         behaviour_args: Optional[dict]=None) -> str:
         """
         Either get the provider info from configuration or through dynamic
         discovery.
 
+        :param behaviour_args:
         :param client: A Client instance
         :param state: A key by which the state of the session can be
             retrieved
@@ -205,7 +209,7 @@ def do_provider_info(self, client=None, state='', behaviour_args=None):
 
         _context = client.client_get("service_context")
         if not _context.get('provider_info'):
-            dynamic_provider_info_discovery(client)
+            dynamic_provider_info_discovery(client, behaviour_args=behaviour_args)
             return _context.get('provider_info')['issuer']
         else:
             _pi = _context.get('provider_info')
@@ -280,16 +284,9 @@ def do_client_registration(self, client=None,
                 _params = RegistrationRequest().parameters()
                 request_args.update({k: v for k, v in behaviour_args.items() if k in _params})
 
-            # _ignore = [k for k in list(request_args.keys()) if k in CALLBACK_URIS]
-            # if _context.get('redirect_uris'):
-            #     if 'redirect_uris' not in _ignore:
-            #         _ignore.append('redirect_uris')
-            #
-            # add_callbacks(_context, _ignore)
-
             load_registration_response(client, request_args=request_args)
 
-    def do_webfinger(self, user):
+    def do_webfinger(self, user: str) -> Client:
         """
         Does OpenID Provider Issuer discovery using webfinger.
 
@@ -304,7 +301,10 @@ def do_webfinger(self, user):
         temporary_client.do_request('webfinger', resource=user)
         return temporary_client
 
-    def client_setup(self, iss_id='', user='', behaviour_args=None):
+    def client_setup(self,
+                     iss_id: Optional[str] = '',
+                     user: Optional[str] = '',
+                     behaviour_args: Optional[dict] = None) -> Client:
         """
         First if no issuer ID is given then the identifier for the user is
         used by the webfinger service to try to find the issuer ID.
@@ -358,11 +358,17 @@ def _get_response_type(self, context, req_args: Optional[dict] = None):
         else:
             return context.get('behaviour')['response_types'][0]
 
-    def init_authorization(self, client=None, state='', req_args=None, behaviour_args=None):
+    def init_authorization(self,
+                           client: Optional[Client] = None,
+                           state: Optional[str] = '',
+                           req_args: Optional[dict] = None,
+                           behaviour_args: Optional[dict] = None) -> dict:
         """
         Constructs the URL that will redirect the user to the authorization
         endpoint of the OP/AS.
 
+        :param behaviour_args:
+        :param state:
         :param client: A Client instance
         :param req_args: Non-default Request arguments
         :return: A dictionary with 2 keys: **url** The authorization redirect
@@ -607,8 +613,7 @@ def userinfo_in_id_token(id_token):
         :param id_token: An :py:class:`oidcmsg.oidc.IDToken` instance
         :return: A dictionary with user information
         """
-        res = dict([(k, id_token[k]) for k in OpenIDSchema.c_param.keys() if
-                    k in id_token])
+        res = dict([(k, id_token[k]) for k in OpenIDSchema.c_param.keys() if k in id_token])
         res.update(id_token.extra())
         return res
 
@@ -629,7 +634,8 @@ def finalize_auth(self, client, issuer: str, response: dict,
 
         _srv = client.get_service('authorization')
         try:
-            authorization_response = _srv.parse_response(response, sformat='dict')
+            authorization_response = _srv.parse_response(response, sformat='dict',
+                                                         behaviour_args=behaviour_args)
         except Exception as err:
             logger.error('Parsing authorization_response: {}'.format(err))
             message = traceback.format_exception(*sys.exc_info())

src/oidcrp/util.py

@@ -9,6 +9,7 @@
 import ssl
 import string
 import sys
+from typing import Optional
 from urllib.parse import parse_qs
 from urllib.parse import urlsplit
 from urllib.parse import urlunsplit
@@ -553,26 +554,3 @@ def load_registration_response(client, request_args=None):
         else:
             if 'error' in response:
                 raise OidcServiceError(response.to_json())
-
-
-def dynamic_provider_info_discovery(client):
-    """
-    This is about performing dynamic Provider Info discovery
-
-    :param client: A :py:class:`oidcrp.oidc.Client` instance
-    """
-    try:
-        client.get_service('provider_info')
-    except KeyError:
-        raise ConfigurationError(
-            'Can not do dynamic provider info discovery')
-    else:
-        _context = client.client_get("service_context")
-        try:
-            _context.set('issuer', _context.config['srv_discovery_url'])
-        except KeyError:
-            pass
-
-        response = client.do_request('provider_info')
-        if is_error_message(response):
-            raise OidcServiceError(response['error'])