Merge branch 'token' into develop2

Author: peppelinux

src/oidcop/authz/__init__.py

@@ -30,9 +30,7 @@ def usage_rules(self, client_id: Optional[str] = ""):
             return _usage_rules
 
         try:
-            _per_client = self.server_get("endpoint_context").cdb[client_id][
-                "token_usage_rules"
-            ]
+            _per_client = self.server_get("endpoint_context").cdb[client_id]["token_usage_rules"]
         except KeyError:
             pass
         else:
@@ -59,14 +57,11 @@ def usage_rules_for(self, client_id, token_type):
             return {}
 
     def __call__(
-        self,
-        session_id: str,
-        request: Union[dict, Message],
-        resources: Optional[list] = None,
+        self, session_id: str, request: Union[dict, Message], resources: Optional[list] = None,
     ) -> Grant:
-        session_info = self.server_get(
-            "endpoint_context"
-        ).session_manager.get_session_info(session_id=session_id, grant=True)
+        session_info = self.server_get("endpoint_context").session_manager.get_session_info(
+            session_id=session_id, grant=True
+        )
         grant = session_info["grant"]
 
         args = self.grant_config.copy()
@@ -87,24 +82,19 @@ def __call__(
         # After this is where user consent should be handled
         scopes = request.get("scope", [])
         grant.scope = scopes
-        grant.claims = self.server_get(
-            "endpoint_context"
-        ).claims_interface.get_claims_all_usage(session_id=session_id, scopes=scopes)
+        grant.claims = self.server_get("endpoint_context").claims_interface.get_claims_all_usage(
+            session_id=session_id, scopes=scopes
+        )
 
         return grant
 
 
 class Implicit(AuthzHandling):
     def __call__(
-        self,
-        session_id: str,
-        request: Union[dict, Message],
-        resources: Optional[list] = None,
+        self, session_id: str, request: Union[dict, Message], resources: Optional[list] = None,
     ) -> Grant:
         args = self.grant_config.copy()
-        grant = self.server_get("endpoint_context").session_manager.get_grant(
-            session_id=session_id
-        )
+        grant = self.server_get("endpoint_context").session_manager.get_grant(session_id=session_id)
         for arg, val in args:
             setattr(grant, arg, val)
         return grant

src/oidcop/client_authn.py

@@ -131,9 +131,7 @@ def is_usable(self, request=None, authorization_token=None):
 
     def verify(self, request, **kwargs):
         if (
-            self.server_get("endpoint_context").cdb[request["client_id"]][
-                "client_secret"
-            ]
+            self.server_get("endpoint_context").cdb[request["client_id"]]["client_secret"]
             == request["client_secret"]
         ):
             return {"client_id": request["client_id"]}
@@ -148,9 +146,7 @@ class BearerHeader(ClientSecretBasic):
     tag = "bearer_header"
 
     def is_usable(self, request=None, authorization_token=None):
-        if authorization_token is not None and authorization_token.startswith(
-            "Bearer "
-        ):
+        if authorization_token is not None and authorization_token.startswith("Bearer "):
             return True
         return False
 
@@ -203,9 +199,7 @@ def verify(self, request, key_type, **kwargs):
         if _sign_alg and _sign_alg.startswith("HS"):
             if key_type == "private_key":
                 raise AttributeError("Wrong key type")
-            keys = _context.keyjar.get(
-                "sig", "oct", ca_jwt["iss"], ca_jwt.jws_header.get("kid")
-            )
+            keys = _context.keyjar.get("sig", "oct", ca_jwt["iss"], ca_jwt.jws_header.get("kid"))
             _secret = _context.cdb[ca_jwt["iss"]].get("client_secret")
             if _secret and keys[0].key != as_bytes(_secret):
                 raise AttributeError("Oct key used for signing not client_secret")
@@ -361,14 +355,10 @@ def verify_client(
         if _method.is_usable(request, authorization_token):
             try:
                 auth_info = _method.verify(
-                    request=request,
-                    authorization_token=authorization_token,
-                    endpoint=endpoint,
+                    request=request, authorization_token=authorization_token, endpoint=endpoint,
                 )
             except Exception as err:
-                logger.warning(
-                    "Verifying auth using {} failed: {}".format(_method.tag, err)
-                )
+                logger.warning("Verifying auth using {} failed: {}".format(_method.tag, err))
             else:
                 if "method" not in auth_info:
                     auth_info["method"] = _method.tag
@@ -398,19 +388,15 @@ def verify_client(
                 raise UnknownClient("Unknown Client ID")
 
         if not valid_client_info(_cinfo):
-            logger.warning(
-                "Client registration has timed out or " "client secret is expired."
-            )
+            logger.warning("Client registration has timed out or " "client secret is expired.")
             raise InvalidClient("Not valid client")
 
         # store what authn method was used
         if auth_info.get("method"):
             _request_type = request.__class__.__name__
             _used_authn_method = endpoint_context.cdb[client_id].get("auth_method")
             if _used_authn_method:
-                endpoint_context.cdb[client_id]["auth_method"][
-                    _request_type
-                ] = auth_info["method"]
+                endpoint_context.cdb[client_id]["auth_method"][_request_type] = auth_info["method"]
             else:
                 endpoint_context.cdb[client_id]["auth_method"] = {
                     _request_type: auth_info["method"]
@@ -422,9 +408,7 @@ def verify_client(
 
         try:
             # get_client_id_from_token is a callback... Do not abuse for code readability.
-            auth_info["client_id"] = get_client_id_from_token(
-                endpoint_context, _token, request
-            )
+            auth_info["client_id"] = get_client_id_from_token(endpoint_context, _token, request)
         except KeyError:
             raise ValueError("Unknown token")