Feature introspection claims

src/oidcendpoint/oauth2/introspection.py

@@ -22,6 +22,9 @@ class Introspection(Endpoint):
     def __init__(self, **kwargs):
         Endpoint.__init__(self, **kwargs)
         self.offset = kwargs.get("offset", 0)
+        self.enable_claims_per_client = kwargs.get(
+            "enable_claims_per_client", False
+        )
 
     def get_client_id_from_token(self, endpoint_context, token, request=None):
         """
@@ -35,6 +38,27 @@ def get_client_id_from_token(self, endpoint_context, token, request=None):
         sinfo = endpoint_context.sdb[token]
         return sinfo["authn_req"]["client_id"]
 
+    def _get_client_claims(self, token):
+        client_id = self.get_client_id_from_token(
+            self.endpoint_context, token
+        )
+        client = self.endpoint_context.cdb.get(client_id, {})
+        return client.get("introspection_claims")
+
+    def _get_user_info(self, token_info):
+        user_id = self.endpoint_context.sdb.sso_db.get_uid_by_sid(
+            token_info["sid"]
+        )
+        return self.endpoint_context.userinfo(user_id, client_id=None)
+
+    def _add_claims(self, token_info, claims, payload):
+        user_info = self._get_user_info(token_info)
+        for attr in claims:
+            try:
+                payload[attr] = user_info[attr]
+            except KeyError:
+                pass
+
     def _introspect(self, token):
         try:
             info = self.endpoint_context.sdb[token]
@@ -89,6 +113,12 @@ def process_request(self, request=None, **kwargs):
 
         _resp.update(_info)
         _resp.weed()
+
+        if self.enable_claims_per_client:
+            client_claims = self._get_client_claims(_token)
+            if client_claims:
+                self._add_claims(_info, client_claims, _resp)
+
         _resp["active"] = True
 
         return {"response_args": _resp}

tests/test_31_introspection.py

@@ -134,6 +134,7 @@ def create_endpoint(self, jwt_token):
                     "kwargs": {
                         "release": ["username"],
                         "client_authn_method": ["client_secret_post"],
+                        "enable_claims_per_client": False,
                     },
                 },
                 "token": {
@@ -176,6 +177,7 @@ 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", "eduperson_scoped_affiliation"],
         }
         endpoint_context.keyjar.import_jwks_as_json(
             endpoint_context.keyjar.export_jwks_as_json(private=True),
@@ -306,6 +308,25 @@ def test_access_token(self):
         assert _resp_args["active"]
         assert _resp_args["scope"] == "openid"
 
+    def test_introspection_claims(self):
+        self.introspection_endpoint.enable_claims_per_client = True
+        _context = self.introspection_endpoint.endpoint_context
+        _token = self._create_at("diana", lifetime=6000, with_jti=True)
+        _req = self.introspection_endpoint.parse_request(
+            {
+                "token": _token,
+                "client_id": "client_1",
+                "client_secret": _context.cdb["client_1"]["client_secret"],
+            }
+        )
+        _resp = self.introspection_endpoint.process_request(_req)
+        _resp_args = _resp["response_args"]
+        assert "nickname" in _resp_args
+        assert _resp_args["nickname"] == 'Dina'
+        assert "eduperson_scoped_affiliation" in _resp_args
+        assert _resp_args["eduperson_scoped_affiliation"] == ['staff@example.org']
+        assert "family_name" not in _resp_args
+
     def test_jwt_unknown_key(self):
         _keyjar = build_keyjar(KEYDEFS)