Merge pull request #35 from IdentityPython/def_req_token_type

Defined DEFAULT_REQUESTED_TOKEN_TYPE.

Author: rohe

src/idpyoidc/server/constant.py

@@ -1,5 +1,3 @@
-import os
-
 # from cryptojwt.jwe.fernet import DEFAULT_ITERATIONS
 
 DEF_SIGN_ALG = {
@@ -17,3 +15,5 @@
 DIVIDER = ";;"
 
 DEFAULT_TOKEN_LIFETIME = 1800
+
+DEFAULT_REQUESTED_TOKEN_TYPE = "urn:ietf:params:oauth:token-type:access_token"

src/idpyoidc/server/oauth2/token.py

@@ -9,6 +9,7 @@
 from idpyoidc.message.oauth2 import ResponseMessage
 from idpyoidc.message.oauth2 import TokenExchangeRequest
 from idpyoidc.message.oidc import TokenErrorResponse
+from idpyoidc.server.constant import DEFAULT_REQUESTED_TOKEN_TYPE
 from idpyoidc.server.endpoint import Endpoint
 from idpyoidc.server.exception import ProcessError
 from idpyoidc.server.oauth2.token_helper import AccessTokenHelper
@@ -133,13 +134,24 @@ def process_request(self, request: Optional[Union[Message, dict]] = None, **kwar
 
         if isinstance(request, TokenExchangeRequest):
             if "token_exchange" in _context.cdb[request["client_id"]]:
-                default_requested_token_type = _context.cdb[request["client_id"]]["token_exchange"][
-                    "default_requested_token_type"
-                ]
+                try:
+                    default_requested_token_type = _context.cdb[request["client_id"]][
+                        "token_exchange"]["default_requested_token_type"]
+                except KeyError:
+                    try:
+                        default_requested_token_type = self.helper[
+                            "urn:ietf:params:oauth:grant-type:token-exchange"
+                        ].config["default_requested_token_type"]
+                    except:
+                        default_requested_token_type = DEFAULT_REQUESTED_TOKEN_TYPE
             else:
-                default_requested_token_type = self.helper[
-                    "urn:ietf:params:oauth:grant-type:token-exchange"
-                ].config["default_requested_token_type"]
+                try:
+                    default_requested_token_type = self.helper[
+                        "urn:ietf:params:oauth:grant-type:token-exchange"
+                    ].config["default_requested_token_type"]
+                except KeyError:
+                    default_requested_token_type = DEFAULT_REQUESTED_TOKEN_TYPE
+
             requested_token_type = request.get("requested_token_type", default_requested_token_type)
             _handler_key = TOKEN_TYPES_MAPPING[requested_token_type]
         else:

src/idpyoidc/server/oauth2/token_helper.py

@@ -13,6 +13,7 @@
 from idpyoidc.message.oauth2 import TokenExchangeResponse
 from idpyoidc.message.oidc import RefreshAccessTokenRequest
 from idpyoidc.message.oidc import TokenErrorResponse
+from idpyoidc.server.constant import DEFAULT_REQUESTED_TOKEN_TYPE
 from idpyoidc.server.constant import DEFAULT_TOKEN_LIFETIME
 from idpyoidc.server.exception import ToOld
 from idpyoidc.server.exception import UnAuthorizedClientScope
@@ -597,11 +598,7 @@ def process_request(self, request, **kwargs):
     def _validate_configuration(self, config):
         if "requested_token_types_supported" not in config:
             raise ImproperlyConfigured(
-                f"Missing 'requested_token_types_supported'" "from Token Exchange configuration"
-            )
-        if "default_requested_token_type" not in config:
-            raise ImproperlyConfigured(
-                f"Missing 'default_requested_token_type'" "from Token Exchange configuration"
+                f"Missing 'requested_token_types_supported' from Token Exchange configuration"
             )
         if "policy" not in config:
             raise ImproperlyConfigured(f"Missing 'policy' from Token Exchange configuration")
@@ -613,11 +610,14 @@ def _validate_configuration(self, config):
             raise ImproperlyConfigured(
                 f"Missing 'callable' from default Token Exchange policy configuration"
             )
-        if config["default_requested_token_type"] not in config["requested_token_types_supported"]:
+
+        _default_requested_token_type = config.get("default_requested_token_type",
+                                                   DEFAULT_REQUESTED_TOKEN_TYPE)
+        if _default_requested_token_type not in config["requested_token_types_supported"]:
             raise ImproperlyConfigured(
-                f"Unsupported default requested_token_type {config['default_requested_token_type']}"
+                f"Unsupported default requested_token_type {_default_requested_token_type}"
             )
-        
+
 
 def validate_token_exchange_policy(request, context, subject_token, **kwargs):
     if "resource" in request: