Merge pull request #74 from IdentityPython/token_type

Implemented our agreement on terms for tokens

Author: rohe

src/oidcop/oauth2/authorization.py

@@ -301,12 +301,12 @@ def verify_response_type(self, request: Union[Message, dict], cinfo: dict) -> bo
         # Is the asked for response_type among those that are permitted
         return set(request["response_type"]) in _registered
 
-    def mint_token(self, token_type, grant, session_id, based_on=None, **kwargs):
-        usage_rules = grant.usage_rules.get(token_type, {})
+    def mint_token(self, token_class, grant, session_id, based_on=None, **kwargs):
+        usage_rules = grant.usage_rules.get(token_class, {})
         token = grant.mint_token(
             session_id=session_id,
             endpoint_context=self.server_get("endpoint_context"),
-            token_type=token_type,
+            token_class=token_class,
             based_on=based_on,
             usage_rules=usage_rules,
             **kwargs,
@@ -677,7 +677,7 @@ def create_authn_response(self, request: Union[dict, Message], sid: str) -> dict
 
             if "code" in request["response_type"]:
                 _code = self.mint_token(
-                    token_type="authorization_code", grant=grant, session_id=_sinfo["session_id"],
+                    token_class="authorization_code", grant=grant, session_id=_sinfo["session_id"],
                 )
                 aresp["code"] = _code.value
                 handled_response_type.append("code")
@@ -686,7 +686,7 @@ def create_authn_response(self, request: Union[dict, Message], sid: str) -> dict
 
             if "token" in rtype:
                 _access_token = self.mint_token(
-                    token_type="access_token", grant=grant, session_id=_sinfo["session_id"],
+                    token_class="access_token", grant=grant, session_id=_sinfo["session_id"],
                 )
                 aresp["access_token"] = _access_token.value
                 aresp["token_type"] = "Bearer"
@@ -707,7 +707,7 @@ def create_authn_response(self, request: Union[dict, Message], sid: str) -> dict
 
                 try:
                     id_token = self.mint_token(
-                        token_type="id_token",
+                        token_class="id_token",
                         grant=grant,
                         session_id=_sinfo["session_id"],
                         **kwargs,

src/oidcop/oauth2/introspection.py

@@ -26,7 +26,7 @@ def __init__(self, server_get, **kwargs):
 
     def _introspect(self, token, client_id, grant):
         # Make sure that the token is an access_token or a refresh_token
-        if token.type not in ["access_token", "refresh_token"]:
+        if token.token_class not in ["access_token", "refresh_token"]:
             return None
 
         if not token.is_active():
@@ -47,13 +47,21 @@ def _introspect(self, token, client_id, grant):
             "active": True,
             "scope": " ".join(scope),
             "client_id": client_id,
-            "token_type": token.type,
+            "token_class": token.token_class,
             "exp": token.expires_at,
             "iat": token.issued_at,
             "sub": grant.sub,
             "iss": _context.issuer,
         }
 
+        try:
+            _token_type = token.token_type
+        except AttributeError:
+            _token_type = None
+
+        if _token_type:
+            ret["token_type"] = _token_type
+
         if aud:
             ret["aud"] = aud
 

src/oidcop/oauth2/token.py

@@ -48,12 +48,13 @@ def process_request(self, req: Union[Message, dict], **kwargs):
 
     def _mint_token(
         self,
-        type: str,
+        token_class: str,
         grant: Grant,
         session_id: str,
         client_id: str,
         based_on: Optional[SessionToken] = None,
         token_args: Optional[dict] = None,
+        token_type: Optional[str] = ""
     ) -> SessionToken:
         _context = self.endpoint.server_get("endpoint_context")
         _mngr = _context.session_manager
@@ -75,10 +76,11 @@ def _mint_token(
         token = grant.mint_token(
             session_id,
             endpoint_context=_context,
-            token_type=type,
-            token_handler=_mngr.token_handler[type],
+            token_class=token_class,
+            token_handler=_mngr.token_handler[token_class],
             based_on=based_on,
             usage_rules=usage_rules,
+            token_type=token_type,
             **_args,
         )
 
@@ -143,7 +145,7 @@ def process_request(self, req: Union[Message, dict], **kwargs):
         if "access_token" in _supports_minting:
             try:
                 token = self._mint_token(
-                    type="access_token",
+                    token_class="access_token",
                     grant=grant,
                     session_id=_session_info["session_id"],
                     client_id=_session_info["client_id"],
@@ -159,7 +161,7 @@ def process_request(self, req: Union[Message, dict], **kwargs):
         if issue_refresh and "refresh_token" in _supports_minting:
             try:
                 refresh_token = self._mint_token(
-                    type="refresh_token",
+                    token_class="refresh_token",
                     grant=grant,
                     session_id=_session_info["session_id"],
                     client_id=_session_info["client_id"],
@@ -227,7 +229,7 @@ def process_request(self, req: Union[Message, dict], **kwargs):
         _grant = _session_info["grant"]
         token = _grant.get_token(token_value)
         access_token = self._mint_token(
-            type="access_token",
+            token_class="access_token",
             grant=_grant,
             session_id=_session_info["session_id"],
             client_id=_session_info["client_id"],
@@ -236,7 +238,7 @@ def process_request(self, req: Union[Message, dict], **kwargs):
 
         _resp = {
             "access_token": access_token.value,
-            "token_type": access_token.type,
+            "token_type": access_token.token_type,
             "scope": _grant.scope,
         }
 
@@ -246,7 +248,7 @@ def process_request(self, req: Union[Message, dict], **kwargs):
         _mints = token.usage_rules.get("supports_minting")
         if "refresh_token" in _mints:
             refresh_token = self._mint_token(
-                type="refresh_token",
+                token_class="refresh_token",
                 grant=_grant,
                 session_id=_session_info["session_id"],
                 client_id=_session_info["client_id"],

src/oidcop/oidc/token.py

@@ -89,11 +89,12 @@ def process_request(self, req: Union[Message, dict], **kwargs):
         if "access_token" in _supports_minting:
             try:
                 token = self._mint_token(
-                    type="access_token",
+                    token_class="access_token",
                     grant=grant,
                     session_id=_session_info["session_id"],
                     client_id=_session_info["client_id"],
                     based_on=_based_on,
+                    token_type=token_type,
                 )
             except MintingNotAllowed as err:
                 logger.warning(err)
@@ -105,7 +106,7 @@ def process_request(self, req: Union[Message, dict], **kwargs):
         if issue_refresh and "refresh_token" in _supports_minting:
             try:
                 refresh_token = self._mint_token(
-                    type="refresh_token",
+                    token_class="refresh_token",
                     grant=grant,
                     session_id=_session_info["session_id"],
                     client_id=_session_info["client_id"],
@@ -123,7 +124,7 @@ def process_request(self, req: Union[Message, dict], **kwargs):
             if "id_token" in _based_on.usage_rules.get("supports_minting"):
                 try:
                     _idtoken = self._mint_token(
-                        type="id_token",
+                        token_class="id_token",
                         grant=grant,
                         session_id=_session_info["session_id"],
                         client_id=_session_info["client_id"],
@@ -202,11 +203,12 @@ def process_request(self, req: Union[Message, dict], **kwargs):
 
         token = _grant.get_token(token_value)
         access_token = self._mint_token(
-            type="access_token",
+            token_class="access_token",
             grant=_grant,
             session_id=_session_info["session_id"],
             client_id=_session_info["client_id"],
             based_on=token,
+            token_type=token_type
         )
 
         _resp = {
@@ -221,7 +223,7 @@ def process_request(self, req: Union[Message, dict], **kwargs):
         _mints = token.usage_rules.get("supports_minting")
         if "refresh_token" in _mints:
             refresh_token = self._mint_token(
-                type="refresh_token",
+                token_class="refresh_token",
                 grant=_grant,
                 session_id=_session_info["session_id"],
                 client_id=_session_info["client_id"],
@@ -233,7 +235,7 @@ def process_request(self, req: Union[Message, dict], **kwargs):
         if "id_token" in _mints:
             try:
                 _idtoken = self._mint_token(
-                    type="refresh_token",
+                    token_class="refresh_token",
                     grant=_grant,
                     session_id=_session_info["session_id"],
                     client_id=_session_info["client_id"],

src/oidcop/oidc/userinfo.py

@@ -110,7 +110,7 @@ def process_request(self, request=None, **kwargs):
         _grant = _session_info["grant"]
         token = _grant.get_token(request["access_token"])
         # should be an access token
-        if token.type != "access_token":
+        if token.token_class != "access_token":
             return self.error_cls(error="invalid_token", error_description="Wrong type of token")
 
         # And it should be valid

src/oidcop/session/claims.py

@@ -25,15 +25,17 @@ def available_claims(endpoint_context):
 
 class ClaimsInterface:
     init_args = {"add_claims_by_scope": False, "enable_claims_per_client": False}
-    claims_types = ["userinfo", "introspection", "id_token", "access_token"]
+    claims_release_points = ["userinfo", "introspection", "id_token", "access_token"]
 
     def __init__(self, server_get):
         self.server_get = server_get
 
-    def authorization_request_claims(self, session_id: str, usage: Optional[str] = "") -> dict:
+    def authorization_request_claims(self,
+                                     session_id: str,
+                                     claims_release_point: Optional[str] = "") -> dict:
         _grant = self.server_get("endpoint_context").session_manager.get_grant(session_id)
         if _grant.authorization_request and "claims" in _grant.authorization_request:
-            return _grant.authorization_request["claims"].get(usage, {})
+            return _grant.authorization_request["claims"].get(claims_release_point, {})
 
         return {}
 
@@ -63,19 +65,19 @@ def _get_module(self, usage, endpoint_context):
 
         return module
 
-    def get_claims(self, session_id: str, scopes: str, usage: str) -> dict:
+    def get_claims(self, session_id: str, scopes: str, claims_release_point: str) -> dict:
         """
 
         :param session_id: Session identifier
         :param scopes: Scopes
-        :param usage: Where to use the claims. One of
-        "userinfo"/"id_token"/"introspection"/"access_token"
+        :param claims_release_point: Where to release the claims. One of
+            "userinfo"/"id_token"/"introspection"/"access_token"
         :return: Claims specification as a dictionary.
         """
 
         _context = self.server_get("endpoint_context")
         # which endpoint module configuration to get the base claims from
-        module = self._get_module(usage, _context)
+        module = self._get_module(claims_release_point, _context)
 
         if module:
             base_claims = module.kwargs.get("base_claims", {})
@@ -86,7 +88,7 @@ def get_claims(self, session_id: str, scopes: str, usage: str) -> dict:
 
         # Can there be per client specification of which claims to use.
         if module.kwargs.get("enable_claims_per_client"):
-            claims = self._get_client_claims(client_id, usage)
+            claims = self._get_client_claims(client_id, claims_release_point)
         else:
             claims = {}
 
@@ -102,7 +104,8 @@ def get_claims(self, session_id: str, scopes: str, usage: str) -> dict:
 
         # Bring in claims specification from the authorization request
         # This only goes for ID Token and user info
-        request_claims = self.authorization_request_claims(session_id=session_id, usage=usage)
+        request_claims = self.authorization_request_claims(session_id=session_id,
+                                                           claims_release_point=claims_release_point)
 
         # This will add claims that has not be added before and
         # set filters on those claims that also appears in one of the sources above
@@ -113,7 +116,7 @@ def get_claims(self, session_id: str, scopes: str, usage: str) -> dict:
 
     def get_claims_all_usage(self, session_id: str, scopes: str) -> dict:
         _claims = {}
-        for usage in self.claims_types:
+        for usage in self.claims_release_points:
             _claims[usage] = self.get_claims(session_id, scopes, usage)
         return _claims
 
@@ -189,7 +192,7 @@ def by_schema(cls, **kwa):
 
 
 class OAuth2ClaimsInterface(ClaimsInterface):
-    claims_types = ["introspection", "access_token"]
+    claims_release_points = ["introspection", "access_token"]
 
     def _get_module(self, usage, endpoint_context):
         module = None
@@ -205,6 +208,6 @@ def _get_module(self, usage, endpoint_context):
 
     def get_claims_all_usage(self, session_id: str, scopes: str) -> dict:
         _claims = {}
-        for usage in self.claims_types:
+        for usage in self.claims_release_points:
             _claims[usage] = self.get_claims(session_id, scopes, usage)
         return _claims