Merge pull request #124 from nsklikas/feature-grant-types-per-client

Add grant_types_supported per client

Author: peppelinux

docs/source/contents/conf.rst

@@ -712,3 +712,11 @@ How to configure the release of the user claims per clients::
                 "id_token": False,
             },
         },
+
+Some of the allowed client configurations are (this is an ongoing work):
+
+---------------------
+grant_types_supported
+---------------------
+
+Configure the allowed grant types on the token endpoint.

src/oidcop/oauth2/token.py

@@ -4,7 +4,6 @@
 
 from cryptojwt.jwe.exception import JWEException
 from cryptojwt.jwt import utc_time_sans_frac
-
 from oidcmsg.message import Message
 from oidcmsg.oauth2 import AccessTokenResponse
 from oidcmsg.oauth2 import ResponseMessage
@@ -390,13 +389,20 @@ def configure_grant_types(self, grant_types_supported):
     def _post_parse_request(
         self, request: Union[Message, dict], client_id: Optional[str] = "", **kwargs
     ):
-        _helper = self.helper.get(request["grant_type"])
+        grant_type = request["grant_type"]
+        _helper = self.helper.get(grant_type)
+        client = kwargs["endpoint_context"].cdb[client_id]
+        if "grant_types_supported" in client and grant_type not in client["grant_types_supported"]:
+            return self.error_cls(
+                error="invalid_request",
+                error_description=f"Unsupported grant_type: {grant_type}",
+            )
         if _helper:
             return _helper.post_parse_request(request, client_id, **kwargs)
         else:
             return self.error_cls(
                 error="invalid_request",
-                error_description=f"Unsupported grant_type: {request['grant_type']}",
+                error_description=f"Unsupported grant_type: {grant_type}",
             )
 
     def process_request(self, request: Optional[Union[Message, dict]] = None, **kwargs):

tests/test_24_oauth2_token_endpoint.py

@@ -234,6 +234,28 @@ def test_parse(self):
 
         assert set(_req.keys()) == set(_token_request.keys())
 
+    def test_auth_code_grant_disallowed_per_client(self):
+        areq = AUTH_REQ.copy()
+        areq["scope"] = ["email"]
+        self.endpoint_context.cdb["client_1"]["grant_types_supported"] = []
+
+        session_id = self._create_session(areq)
+        grant = self.endpoint_context.authz(session_id, areq)
+        code = self._mint_code(grant, areq["client_id"])
+
+        _cntx = self.endpoint_context
+
+        _token_request = TOKEN_REQ_DICT.copy()
+        _token_request["code"] = code.value
+        _req = self.token_endpoint.parse_request(_token_request)
+        _resp = self.token_endpoint.process_request(request=_req, issue_refresh=True)
+
+        assert isinstance(_req, TokenErrorResponse)
+        assert _req.to_dict() == {
+            "error": "invalid_request",
+            "error_description": "Unsupported grant_type: authorization_code",
+        }
+
     def test_process_request(self):
         session_id = self._create_session(AUTH_REQ)
         grant = self.session_manager[session_id]
@@ -336,6 +358,40 @@ def test_do_refresh_access_token(self):
         msg = self.token_endpoint.do_response(request=_req, **_resp)
         assert isinstance(msg, dict)
 
+    def test_refresh_grant_disallowed_per_client(self):
+        areq = AUTH_REQ.copy()
+        areq["scope"] = ["email"]
+        self.endpoint_context.cdb["client_1"]["grant_types_supported"] = [
+            "authorization_code"
+        ]
+
+        session_id = self._create_session(areq)
+        grant = self.endpoint_context.authz(session_id, areq)
+        code = self._mint_code(grant, areq["client_id"])
+
+        _cntx = self.endpoint_context
+
+        _token_request = TOKEN_REQ_DICT.copy()
+        _token_request["code"] = code.value
+        _req = self.token_endpoint.parse_request(_token_request)
+        _resp = self.token_endpoint.process_request(request=_req, issue_refresh=True)
+
+        _request = REFRESH_TOKEN_REQ.copy()
+        _request["refresh_token"] = _resp["response_args"]["refresh_token"]
+
+        _token_value = _resp["response_args"]["refresh_token"]
+        _session_info = self.session_manager.get_session_info_by_token(_token_value)
+        _token = self.session_manager.find_token(_session_info["session_id"], _token_value)
+        _token.usage_rules["supports_minting"] = ["access_token", "refresh_token"]
+
+        _req = self.token_endpoint.parse_request(_request.to_json())
+
+        assert isinstance(_req, TokenErrorResponse)
+        assert _req.to_dict() == {
+            "error": "invalid_request",
+            "error_description": "Unsupported grant_type: refresh_token",
+        }
+
     def test_do_2nd_refresh_access_token(self):
         areq = AUTH_REQ.copy()
         areq["scope"] = ["email"]

tests/test_35_oidc_token_endpoint.py

@@ -29,6 +29,7 @@
 from oidcop.user_authn.authn_context import INTERNETPROTOCOLPASSWORD
 from oidcop.user_info import UserInfo
 from oidcop.util import lv_pack
+from tests.test_24_oauth2_token_endpoint import TestEndpoint as _TestEndpoint
 
 KEYDEFS = [
     {"type": "RSA", "key": "", "use": ["sig"]},
@@ -183,7 +184,7 @@ def conf():
     }
 
 
-class TestEndpoint(object):
+class TestEndpoint(_TestEndpoint):
     @pytest.fixture(autouse=True)
     def create_endpoint(self, conf):
         server = Server(OPConfiguration(conf=conf, base_path=BASEDIR), cwd=BASEDIR)