Merge remote-tracking branch 'origin/main'

Author: rohe

src/idpyoidc/server/oauth2/authorization.py

@@ -268,7 +268,9 @@ def authn_args_gather(
 
 
 def check_unknown_scopes_policy(request_info, client_id, context):
-    if not context.get_preference("deny_unknown_scopes"):
+    cinfo = context.cdb.get(client_id, {})
+    deny_unknown_scopes = cinfo.get("deny_unknown_scopes", context.get_preference("deny_unknown_scopes"))
+    if not deny_unknown_scopes:
         return
 
     scope = request_info["scope"]

src/idpyoidc/server/oauth2/token.py

@@ -189,4 +189,4 @@ def process_request(self, request: Optional[Union[Message, dict]] = None, **kwar
         return resp
 
     def supports(self):
-        return {"grant_types_supported": list(self.grant_type_helper.keys())}
+        return self._supports

src/idpyoidc/server/oidc/token.py

@@ -26,14 +26,21 @@ class Token(token.Token):
     name = "token"
     default_capabilities = None
 
+    helper_by_grant_type = {
+        "authorization_code": AccessTokenHelper,
+        "refresh_token": RefreshTokenHelper,
+        "urn:ietf:params:oauth:grant-type:token-exchange": TokenExchangeHelper,
+    }
+
     _supports = {
         "token_endpoint_auth_methods_supported": [
             "client_secret_post",
             "client_secret_basic",
             "client_secret_jwt",
             "private_key_jwt",
         ],
-        "token_endpoint_auth_signing_alg_values_supported": claims.get_signing_algs,
+        "token_endpoint_auth_signing_alg_values_supported": claims.get_signing_algs(),
+        "grant_types_supported": list(helper_by_grant_type.keys())
     }
 
     helper_by_grant_type = {

src/idpyoidc/server/oidc/userinfo.py

@@ -133,44 +133,22 @@ def process_request(self, request=None, **kwargs):
         if token.is_active() is False:
             return self.error_cls(error="invalid_token", error_description="Invalid Token")
 
-        allowed = True
-        _auth_event = _grant.authentication_event
-        # if the authentication is still active or offline_access is granted.
-        if not _auth_event["valid_until"] >= utc_time_sans_frac():
-            logger.debug(
-                "authentication not valid: {} > {}".format(
-                    datetime.fromtimestamp(_auth_event["valid_until"]),
-                    datetime.fromtimestamp(utc_time_sans_frac()),
-                )
-            )
-            allowed = False
-
-            # This has to be made more fine grained.
-            # if "offline_access" in session["authn_req"]["scope"]:
-            #     pass
-
-        if allowed:
-            _cntxt = self.upstream_get("context")
-            _claims_restriction = _cntxt.claims_interface.get_claims(
-                _session_info["branch_id"], scopes=token.scope, claims_release_point="userinfo"
-            )
-            info = _cntxt.claims_interface.get_user_claims(
-                _session_info["user_id"], claims_restriction=_claims_restriction
-            )
-            info["sub"] = _grant.sub
-            if _grant.add_acr_value("userinfo"):
-                info["acr"] = _grant.authentication_event["authn_info"]
+        _cntxt = self.upstream_get("context")
+        _claims_restriction = _cntxt.claims_interface.get_claims(
+            _session_info["branch_id"], scopes=token.scope, claims_release_point="userinfo"
+        )
+        info = _cntxt.claims_interface.get_user_claims(
+            _session_info["user_id"], claims_restriction=_claims_restriction
+        )
+        info["sub"] = _grant.sub
+        if _grant.add_acr_value("userinfo"):
+            info["acr"] = _grant.authentication_event["authn_info"]
 
-            if "userinfo" in _cntxt.cdb[request["client_id"]]:
-                self.config["policy"] = _cntxt.cdb[request["client_id"]]["userinfo"]["policy"]
+        if "userinfo" in _cntxt.cdb[request["client_id"]]:
+            self.config["policy"] = _cntxt.cdb[request["client_id"]]["userinfo"]["policy"]
 
-            if "policy" in self.config:
-                info = self._enforce_policy(request, info, token, self.config)
-        else:
-            info = {
-                "error": "invalid_request",
-                "error_description": "Access not granted",
-            }
+        if "policy" in self.config:
+            info = self._enforce_policy(request, info, token, self.config)
 
         return {"response_args": info, "client_id": _session_info["client_id"]}
 

tests/test_server_24_oauth2_authorization_endpoint.py

@@ -588,6 +588,38 @@ def test_setup_auth_invalid_scope(self):
         assert excp
         assert isinstance(excp, UnAuthorizedClientScope)
 
+    def test_setup_auth_invalid_scope_2(self):
+        request = AuthorizationRequest(
+            client_id="client_id",
+            redirect_uri="https://rp.example.com/cb",
+            response_type=["id_token"],
+            state="state",
+            nonce="nonce",
+            scope="openid THAT-BLOODY_SCOPE",
+        )
+        cinfo = {
+            "client_id": "client_id",
+            "redirect_uris": [("https://rp.example.com/cb", {})],
+            "id_token_signed_response_alg": "RS256",
+            "allowed_scopes": ["openid", "profile", "email", "address", "phone", "offline_access"],
+            "deny_unknown_scopes": True
+        }
+
+        _context = self.endpoint.upstream_get("context")
+        _context.cdb["client_id"] = cinfo
+
+        kaka = _context.cookie_handler.make_cookie_content("value", "sso")
+
+        # force to 400 Http Error message if the release scope policy is heavy!
+        _context.set_preference("deny_unknown_scopes", False)
+        excp = None
+        try:
+            res = self.endpoint.process_request(request, http_info={"headers": {"cookie": [kaka]}})
+        except UnAuthorizedClientScope as e:
+            excp = e
+        assert excp
+        assert isinstance(excp, UnAuthorizedClientScope)
+
     def test_setup_auth_user(self):
         request = AuthorizationRequest(
             client_id="client_id",

tests/test_server_26_oidc_userinfo_endpoint.py

@@ -310,24 +310,6 @@ def test_process_request(self):
         args = self.endpoint.process_request(_req, http_info=http_info)
         assert args
 
-    def test_process_request_not_allowed(self):
-        session_id = self._create_session(AUTH_REQ)
-        grant = self.session_manager[session_id]
-        code = self._mint_code(grant, session_id)
-        access_token = self._mint_token("access_token", grant, session_id, code)
-
-        # 2 things can make the request invalid.
-        # 1) The token is not valid anymore or 2) The event is not valid.
-        _event = grant.authentication_event
-        _event["authn_time"] -= 9000
-        _event["valid_until"] -= 9000
-
-        http_info = {"headers": {"authorization": "Bearer {}".format(access_token.value)}}
-        _req = self.endpoint.parse_request({}, http_info=http_info)
-
-        args = self.endpoint.process_request(_req, http_info=http_info)
-        assert set(args["response_args"].keys()) == {"error", "error_description"}
-
     def test_do_response(self):
         session_id = self._create_session(AUTH_REQ)
         grant = self.session_manager[session_id]