WIP: Introduce add_claims_by_scope per client configuration

src/oidcop/session/claims.py

@@ -41,7 +41,11 @@ def authorization_request_claims(self,
 
     def _get_client_claims(self, client_id, usage):
         client_info = self.server_get("endpoint_context").cdb.get(client_id, {})
-        client_claims = client_info.get("{}_claims".format(usage), {})
+        client_claims = (
+            client_info.get("add_claims", {})
+            .get("always", {})
+            .get(usage, {})
+        )
         if isinstance(client_claims, list):
             client_claims = {k: None for k in client_claims}
         return client_claims
@@ -94,8 +98,19 @@ def get_claims(self, session_id: str, scopes: str, claims_release_point: str) ->
 
         claims.update(base_claims)
 
-        # Scopes can in some cases equate to set of claims, is that used here ?
-        if module.kwargs.get("add_claims_by_scope"):
+        # If specific client configuration exists overwrite add_claims_by_scope
+        if client_id in _context.cdb:
+            add_claims_by_scope = (
+                _context.cdb[client_id].get("add_claims", {})
+                .get("by_scope", {})
+                .get(claims_release_point, {})
+            )
+            if isinstance(add_claims_by_scope, dict) and not add_claims_by_scope:
+                add_claims_by_scope = module.kwargs.get("add_claims_by_scope")
+        else:
+            add_claims_by_scope = module.kwargs.get("add_claims_by_scope")
+
+        if add_claims_by_scope:
             if scopes:
                 _scopes = _context.scopes_handler.filter_scopes(client_id, _context, scopes)
 

tests/test_01_claims.py

@@ -114,6 +114,9 @@ def create_idtoken(self):
             "client_salt": "salted",
             "token_endpoint_auth_method": "client_secret_post",
             "response_types": ["code", "token", "code id_token", "id_token"],
+            "add_claims": {
+                "always": {},
+            },
         }
         server.endpoint_context.keyjar.add_symmetric(
             "client_1", "hemligtochintekort", ["sig", "enc"]
@@ -173,17 +176,17 @@ def test_get_client_claims_0(self, usage):
         assert claims == {}
 
     def test_get_client_claims_id_token_1(self):
-        self.endpoint_context.cdb["client_1"]["id_token_claims"] = ["name", "email"]
+        self.endpoint_context.cdb["client_1"]["add_claims"]["always"]["id_token"] = ["name", "email"]
         claims = self.claims_interface._get_client_claims("client_1", "id_token")
         assert set(claims.keys()) == {"name", "email"}
 
     def test_get_client_claims_userinfo_1(self):
-        self.endpoint_context.cdb["client_1"]["userinfo_claims"] = ["email", "address"]
+        self.endpoint_context.cdb["client_1"]["add_claims"]["always"]["userinfo"] = ["email", "address"]
         claims = self.claims_interface._get_client_claims("client_1", "userinfo")
         assert set(claims.keys()) == {"address", "email"}
 
     def test_get_client_claims_introspection_1(self):
-        self.endpoint_context.cdb["client_1"]["introspection_claims"] = ["email"]
+        self.endpoint_context.cdb["client_1"]["add_claims"]["always"]["introspection"] = ["email"]
         claims = self.claims_interface._get_client_claims("client_1", "introspection")
         assert set(claims.keys()) == {"email"}
 
@@ -207,7 +210,7 @@ def test_get_claims_id_token_2(self):
             "base_claims": {"email": None, "email_verified": None},
             "enable_claims_per_client": True,
         }
-        self.endpoint_context.cdb["client_1"]["id_token_claims"] = ["name", "email"]
+        self.endpoint_context.cdb["client_1"]["add_claims"]["always"]["id_token"] = ["name", "email"]
 
         claims = self.claims_interface.get_claims(session_id, [], "id_token")
         assert set(claims.keys()) == {"name", "email", "email_verified"}
@@ -219,7 +222,7 @@ def test_get_claims_id_token_3(self):
             "enable_claims_per_client": True,
             "add_claims_by_scope": True,
         }
-        self.endpoint_context.cdb["client_1"]["id_token_claims"] = ["name", "email"]
+        self.endpoint_context.cdb["client_1"]["add_claims"]["always"]["id_token"] = ["name", "email"]
 
         claims = self.claims_interface.get_claims(session_id, ["openid", "address"], "id_token")
         assert set(claims.keys()) == {
@@ -238,7 +241,7 @@ def test_get_claims_userinfo_3(self):
             "enable_claims_per_client": True,
             "add_claims_by_scope": True,
         }
-        self.endpoint_context.cdb["client_1"]["userinfo_claims"] = ["name", "email"]
+        self.endpoint_context.cdb["client_1"]["add_claims"]["always"]["userinfo"] = ["name", "email"]
 
         claims = self.claims_interface.get_claims(session_id, ["openid", "address"], "userinfo")
         assert set(claims.keys()) == {
@@ -256,7 +259,7 @@ def test_get_claims_introspection_3(self):
             "enable_claims_per_client": True,
             "add_claims_by_scope": True,
         }
-        self.endpoint_context.cdb["client_1"]["introspection_claims"] = [
+        self.endpoint_context.cdb["client_1"]["add_claims"]["always"]["introspection"] = [
             "name",
             "email",
         ]
@@ -280,7 +283,7 @@ def test_get_claims_access_token_3(self):
             "enable_claims_per_client": True,
             "add_claims_by_scope": True,
         }
-        self.endpoint_context.cdb["client_1"]["access_token_claims"] = ["name", "email"]
+        self.endpoint_context.cdb["client_1"]["add_claims"]["always"]["access_token"] = ["name", "email"]
 
         session_id = self._create_session(AREQ)
         claims = self.claims_interface.get_claims(session_id, ["openid", "address"], "access_token")
@@ -320,7 +323,7 @@ def test_get_claims_all_usage_2(self):
         self.server.server_get("endpoint", "userinfo").kwargs = {
             "enable_claims_per_client": True,
         }
-        self.endpoint_context.cdb["client_1"]["userinfo_claims"] = ["name", "email"]
+        self.endpoint_context.cdb["client_1"]["add_claims"]["always"]["userinfo"] = ["name", "email"]
 
         self.server.server_get("endpoint", "introspection").kwargs = {"add_claims_by_scope": True}
 
@@ -349,7 +352,7 @@ def test_get_user_claims(self):
         self.server.server_get("endpoint", "userinfo").kwargs = {
             "enable_claims_per_client": True,
         }
-        self.endpoint_context.cdb["client_1"]["userinfo_claims"] = ["name", "email"]
+        self.endpoint_context.cdb["client_1"]["add_claims"]["always"]["userinfo"] = ["name", "email"]
 
         self.server.server_get("endpoint", "introspection").kwargs = {"add_claims_by_scope": True}
 

tests/test_05_id_token.py

@@ -170,6 +170,10 @@ def create_session_manager(self):
             "client_salt": "salted",
             "token_endpoint_auth_method": "client_secret_post",
             "response_types": ["code", "token", "code id_token", "id_token"],
+            "add_claims": {
+                "always": {},
+                "by_scope": {},
+            },
         }
         self.endpoint_context.keyjar.add_symmetric("client_1", "hemligtochintekort", ["sig", "enc"])
         self.session_manager = self.endpoint_context.session_manager
@@ -487,7 +491,7 @@ def test_client_claims(self):
         grant = self.session_manager[session_id]
 
         self.session_manager.token_handler["id_token"].kwargs["enable_claims_per_client"] = True
-        self.endpoint_context.cdb["client_1"]["id_token_claims"] = {"address": None}
+        self.endpoint_context.cdb["client_1"]["add_claims"]["always"]["id_token"] = {"address": None}
 
         _claims = self.endpoint_context.claims_interface.get_claims(
             session_id=session_id, scopes=AREQ["scope"], claims_release_point="id_token"
@@ -543,6 +547,28 @@ def test_client_claims_scopes(self):
         assert "email" in res
         assert "nickname" not in res
 
+    def test_client_claims_scopes_per_client(self):
+        session_id = self._create_session(AREQS)
+        grant = self.session_manager[session_id]
+        self.session_manager.token_handler["id_token"].kwargs["add_claims_by_scope"] = True
+        self.endpoint_context.cdb[AREQS["client_id"]]["add_claims"]["by_scope"]["id_token"] = False
+
+        _claims = self.endpoint_context.claims_interface.get_claims(
+            session_id=session_id, scopes=AREQS["scope"], claims_release_point="id_token"
+        )
+        grant.claims = {"id_token": _claims}
+
+        id_token = self._mint_id_token(grant, session_id)
+
+        client_keyjar = KeyJar()
+        _jwks = self.endpoint_context.keyjar.export_jwks()
+        client_keyjar.import_jwks(_jwks, self.endpoint_context.issuer)
+        _jwt = JWT(key_jar=client_keyjar, iss="client_1")
+        res = _jwt.unpack(id_token.value)
+        assert "address" not in res
+        assert "email" in res
+        assert "nickname" not in res
+
     def test_client_claims_scopes_and_request_claims_no_match(self):
         session_id = self._create_session(AREQRC)
         grant = self.session_manager[session_id]

tests/test_05_jwt_token.py

@@ -186,6 +186,10 @@ def create_endpoint(self):
             "client_salt": "salted",
             "token_endpoint_auth_method": "client_secret_post",
             "response_types": ["code", "token", "code id_token", "id_token"],
+            "add_claims": {
+                "always": {},
+                "by_scope": {},
+            },
         }
         self.session_manager = self.endpoint_context.session_manager
         self.user_id = "diana"
@@ -247,7 +251,7 @@ def test_info(self):
     @pytest.mark.parametrize("enable_claims_per_client", [True, False])
     def test_enable_claims_per_client(self, enable_claims_per_client):
         # Set up configuration
-        self.endpoint_context.cdb["client_1"]["access_token_claims"] = {"address": None}
+        self.endpoint_context.cdb["client_1"]["add_claims"]["always"]["access_token"] = {"address": None}
         self.endpoint_context.session_manager.token_handler.handler["access_token"].kwargs[
             "enable_claims_per_client"
         ] = enable_claims_per_client

tests/test_07_userinfo.py

@@ -255,7 +255,12 @@ def create_endpoint_context(self):
         server = Server(OPConfiguration(conf=conf, base_path=BASEDIR), cwd=BASEDIR)
         self.endpoint_context = server.endpoint_context
         # Just has to be there
-        self.endpoint_context.cdb["client1"] = {}
+        self.endpoint_context.cdb["client1"] = {
+            "add_claims": {
+                "always": {},
+                "by_scope": {},
+            },
+        }
         self.session_manager = self.endpoint_context.session_manager
         self.claims_interface = ClaimsInterface(server.server_get)
         self.user_id = "diana"
@@ -371,7 +376,7 @@ def test_collect_user_info_enable_claims_per_client(self):
         _userinfo_endpoint.kwargs["enable_claims_per_client"] = True
         del _userinfo_endpoint.kwargs["base_claims"]
 
-        self.endpoint_context.cdb[_req["client_id"]]["userinfo_claims"] = {"phone_number": None}
+        self.endpoint_context.cdb[_req["client_id"]]["add_claims"]["always"]["userinfo"] = {"phone_number": None}
 
         _userinfo_restriction = self.claims_interface.get_claims(
             session_id=session_id, scopes=_req["scope"], claims_release_point="userinfo"

tests/test_31_oauth2_introspection.py

@@ -192,7 +192,12 @@ def create_endpoint(self, jwt_token):
             "client_salt": "salted",
             "token_endpoint_auth_method": "client_secret_post",
             "response_types": ["code", "token", "code id_token", "id_token"],
-            "introspection_claims": {"nickname": None, "eduperson_scoped_affiliation": None,},
+            "add_claims": {
+                "always": {
+                    "introspection": ["nickname", "eduperson_scoped_affiliation"],
+                },
+                "by_scope": {},
+            },
         }
         endpoint_context.keyjar.import_jwks_as_json(
             endpoint_context.keyjar.export_jwks_as_json(private=True), endpoint_context.issuer,