First walk through implementing our agreement on token_type, token_syntax and token_class.

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():
@@ -44,7 +44,7 @@ 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,

src/oidcop/oauth2/token.py

@@ -48,7 +48,7 @@ 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,
@@ -75,8 +75,8 @@ 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,
             **_args,
@@ -143,7 +143,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 +159,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 +227,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 +236,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 +246,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,7 +89,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"],
@@ -105,7 +105,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 +123,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,7 +202,7 @@ 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"],
@@ -221,7 +221,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 +233,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

@@ -30,10 +30,12 @@ class ClaimsInterface:
     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_ref: 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_ref, {})
 
         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_ref: 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_ref: 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_ref, _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_ref)
         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_ref=claims_release_ref)
 
         # 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

src/oidcop/session/grant.py

@@ -41,14 +41,6 @@ def __init__(
         self.resources = resources
 
 
-GRANT_TYPE_MAP = {
-    "authorization_code": "code",
-    "access_token": "access_token",
-    "refresh_token": "refresh_token",
-    "id_token": "id_token",
-}
-
-
 def find_token(issued, token_id):
     for iss in issued:
         if iss.id == token_id:
@@ -179,12 +171,17 @@ def payload_arguments(
             self,
             session_id: str,
             endpoint_context,
-            token_type: str,
+            claims_release_ref: str,
             scope: Optional[dict] = None,
             extra_payload: Optional[dict] = None,
     ) -> dict:
         """
 
+        :param session_id:
+        :param endpoint_context:
+        :param claims_release_ref: One of "userinfo", "introspection", "id_token", "access_token"
+        :param scope:
+        :param extra_payload:
         :return: dictionary containing information to place in a token value
         """
         if not scope:
@@ -205,7 +202,7 @@ def payload_arguments(
                 payload.update({"client_id": client_id, "sub": client_id})
 
         _claims_restriction = endpoint_context.claims_interface.get_claims(
-            session_id, scopes=scope, usage=token_type
+            session_id, scopes=scope, claims_release_ref=claims_release_ref
         )
         user_id, _, _ = endpoint_context.session_manager.decrypt_session_id(session_id)
         user_info = endpoint_context.claims_interface.get_user_claims(user_id, _claims_restriction)
@@ -217,7 +214,7 @@ def mint_token(
             self,
             session_id: str,
             endpoint_context: object,
-            token_type: str,
+            token_class: str,
             token_handler: TokenHandler = None,
             based_on: Optional[SessionToken] = None,
             usage_rules: Optional[dict] = None,
@@ -240,42 +237,46 @@ def mint_token(
             return None
 
         if based_on:
-            if based_on.supports_minting(token_type) is False:
-                raise MintingNotAllowed(f"Minting of {token_type} not supported")
+            if based_on.supports_minting(token_class) is False:
+                raise MintingNotAllowed(f"Minting of {token_class} not supported")
             if not based_on.is_active():
                 raise MintingNotAllowed("Token inactive")
             _base_on_ref = based_on.value
         else:
             _base_on_ref = None
 
-        if usage_rules is None and token_type in self.usage_rules:
-            usage_rules = self.usage_rules[token_type]
+        if usage_rules is None and token_class in self.usage_rules:
+            usage_rules = self.usage_rules[token_class]
 
-        token_class = self.token_map.get(token_type)
-        if token_type == "id_token":
+        _class = self.token_map.get(token_class)
+        if token_class == "id_token":
             class_args = {k: v for k, v in kwargs.items() if k not in ["code", "access_token"]}
             handler_args = {k: v for k, v in kwargs.items() if k in ["code", "access_token"]}
         else:
             class_args = kwargs
             handler_args = {}
 
-        if token_class:
-            item = token_class(
-                type=token_type,
+        if _class:
+            item = _class(
+                token_class=token_class,
                 based_on=_base_on_ref,
                 usage_rules=usage_rules,
                 scope=scope,
                 **class_args,
             )
             if token_handler is None:
-                token_handler = endpoint_context.session_manager.token_handler.handler[
-                    GRANT_TYPE_MAP[token_type]
-                ]
+                token_handler = endpoint_context.session_manager.token_handler.handler[token_class]
+
+            # Only access_token and id_token can give rise to claims release
+            if token_class in ["access_token", "id_token"]:
+                claims_release_ref = token_class
+            else:
+                claims_release_ref = ""
 
             token_payload = self.payload_arguments(
                 session_id,
                 endpoint_context,
-                token_type=token_type,
+                claims_release_ref=claims_release_ref,
                 scope=scope,
                 extra_payload=handler_args,
             )