Find scope

src/oidcop/oauth2/introspection.py

@@ -34,7 +34,10 @@ def _introspect(self, token, client_id, grant):
 
         scope = token.scope
         if not scope:
-            scope = grant.scope
+            if token.based_on:
+                scope = grant.find_scope(token.based_on)
+            else:
+                scope = grant.scope
         aud = token.resources
         if not aud:
             aud = grant.resources

src/oidcop/session/grant.py

@@ -175,6 +175,19 @@ def get(self) -> object:
             resources=self.resources,
         )
 
+    def find_scope(self, based_on):
+        if isinstance(based_on, str):
+            based_on = self.get_token(based_on)
+
+        if based_on:
+            if based_on.scope:
+                return based_on.scope
+
+            if based_on.based_on:
+                return self.find_scope(based_on.based_on)
+
+        return self.scope
+
     def payload_arguments(
             self,
             session_id: str,
@@ -187,7 +200,7 @@ def payload_arguments(
 
         :return: dictionary containing information to place in a token value
         """
-        if not scope:
+        if scope is None:
             scope = self.scope
 
         payload = {"scope": scope, "aud": self.resources, "jti": uuid1().hex}
@@ -260,6 +273,12 @@ def mint_token(
             handler_args = {}
 
         if token_class:
+            if scope is None:
+                if based_on:
+                    scope = self.find_scope(based_on)
+                else:
+                    scope = self.scope
+
             item = token_class(
                 type=token_type,
                 based_on=_base_on_ref,

tests/test_01_grant.py

@@ -438,3 +438,73 @@ def test_get_usage_rules(self):
 
         # client specific usage rules
         self.endpoint_context.cdb["client_id"] = {"access_token": {"expires_in": 600}}
+
+    def test_assigned_scope(self):
+        session_id = self._create_session(AREQ)
+        session_info = self.endpoint_context.session_manager.get_session_info(
+            session_id=session_id, grant=True
+        )
+        grant = session_info["grant"]
+        code = grant.mint_token(
+            session_id,
+            endpoint_context=self.endpoint_context,
+            token_type="authorization_code",
+            token_handler=TOKEN_HANDLER["authorization_code"],
+        )
+
+        code.scope = ["openid", "email"]
+
+        access_token = grant.mint_token(
+            session_id,
+            endpoint_context=self.endpoint_context,
+            token_type="access_token",
+            token_handler=TOKEN_HANDLER["access_token"],
+            based_on=code,
+        )
+
+        assert access_token.scope == code.scope
+
+    def test_assigned_scope_2nd(self):
+        session_id = self._create_session(AREQ)
+        session_info = self.endpoint_context.session_manager.get_session_info(
+            session_id=session_id, grant=True
+        )
+        grant = session_info["grant"]
+        code = grant.mint_token(
+            session_id,
+            endpoint_context=self.endpoint_context,
+            token_type="authorization_code",
+            token_handler=TOKEN_HANDLER["authorization_code"],
+        )
+
+        code.scope = ["openid", "email"]
+
+        refresh_token = grant.mint_token(
+            session_id,
+            endpoint_context=self.endpoint_context,
+            token_type="refresh_token",
+            token_handler=TOKEN_HANDLER["refresh_token"],
+            based_on=code,
+        )
+
+        access_token = grant.mint_token(
+            session_id,
+            endpoint_context=self.endpoint_context,
+            token_type="access_token",
+            token_handler=TOKEN_HANDLER["access_token"],
+            based_on=refresh_token,
+        )
+
+        assert access_token.scope == code.scope
+
+        refresh_token.scope = ["openid", "xyz"]
+
+        access_token = grant.mint_token(
+            session_id,
+            endpoint_context=self.endpoint_context,
+            token_type="access_token",
+            token_handler=TOKEN_HANDLER["access_token"],
+            based_on=refresh_token,
+        )
+
+        assert access_token.scope == refresh_token.scope

tests/test_36_oauth2_token_exchange.py

@@ -197,14 +197,15 @@ def _mint_code(self, grant, session_id):
             token_handler=self.session_manager.token_handler["code"],
         )
 
-    def _mint_access_token(self, grant, session_id, token_ref=None, resources=None):
+    def _mint_access_token(self, grant, session_id, token_ref=None, resources=None, scope=None):
         return grant.mint_token(
             session_id=session_id,
             endpoint_context=self.endpoint.server_get("endpoint_context"),
             token_type="access_token",
             token_handler=self.session_manager.token_handler["access_token"],
             based_on=token_ref,
             resources=resources,
+            scope=scope
         )
 
     def exchange_grant(self, session_id, users, targets, scope):
@@ -257,15 +258,19 @@ def test_do_response(self):
         assert exch_grants
         exch_grant = exch_grants[0]
 
-        session_info = self.session_manager.get_session_info_by_token(ter["subject_token"])
+        session_info = self.session_manager.get_session_info_by_token(ter["subject_token"],
+                                                                      grant=True)
         _token = self.session_manager.find_token(session_info["session_id"], ter["subject_token"])
 
         session_id = self.session_manager.encrypted_session_id(
             session_info["user_id"], session_info["client_id"], exch_grant.id
         )
 
+        _scope = session_info["grant"].find_scope(ter["subject_token"])
+
         _token = self._mint_access_token(
             exch_grant, session_id, token_ref=_token, resources=["https://backend.example.com"],
+            scope=_scope
         )
 
         print(_token.value)
@@ -274,8 +279,7 @@ def test_do_response(self):
                 "token": _token.value,
                 "client_id": "client_1",
                 "client_secret": self.introspection_endpoint.server_get("endpoint_context").cdb[
-                    "client_1"
-                ]["client_secret"],
+                    "client_1"]["client_secret"],
             }
         )
         _resp = self.introspection_endpoint.process_request(_req)