Improve scope matching logic to return the stored token with the minimum number of scopes rather than first match

Author: sam-muncke

src/auth0_server_python/auth_server/server_client.py

@@ -663,7 +663,7 @@ def _merge_scope_with_defaults(
         default_scopes_list = default_scopes.split()
         request_scopes_list = (request_scope or "").split()
 
-        merged_scopes = default_scopes_list + [x for x in request_scopes_list if x not in default_scopes_list]
+        merged_scopes = list(dict.fromkeys(default_scopes_list + request_scopes_list))
         return " ".join(merged_scopes) if merged_scopes else None
 
 
@@ -674,16 +674,20 @@ def _find_matching_token_set(
         scope: Optional[str]
     ) -> Optional[dict[str, Any]]:
         audience = audience or self.DEFAULT_AUDIENCE_STATE_KEY
-        match = None
+        requested_scopes = set(scope.split()) if scope else set()
+        matches: list[tuple[int, dict]] = []
         for token_set in token_sets:
             token_set_audience = token_set.get("audience")
             token_set_scopes = set(token_set.get("scope", "").split())
-            requested_scopes = set(scope.split()) if scope else set()
+            if token_set_audience == audience and token_set_scopes == requested_scopes:
+                # short-circuit if exact match
+                return token_set
             if token_set_audience == audience and token_set_scopes.issuperset(requested_scopes):
-                match = token_set
-                break
+                # consider stored tokens with more scopes than requested by number of scopes
+                matches.append((len(token_set_scopes), token_set))
 
-        return match
+        # Return the token set with the smallest superset of scopes that matches the requested audience and scopes
+        return min(matches, key=lambda t: t[0])[1] if matches else None
 
     async def get_access_token_for_connection(
         self,

src/auth0_server_python/tests/test_server_client.py

@@ -696,6 +696,47 @@ async def test_get_access_token_from_store_with_a_superset_of_requested_scopes(m
     assert token == "other_token_from_store"
     get_refresh_token_mock.assert_not_awaited()
 
+
+@pytest.mark.asyncio
+async def test_get_access_token_from_store_returns_minimum_matching_scopes(mocker):
+    mock_state_store = AsyncMock()
+    mock_state_store.get.return_value = {
+        "refresh_token": None,
+        "token_sets": [
+            {
+                "audience": "some_audience",
+                "access_token": "maximum_scope_token",
+                "scope": "read:foo write:foo read:bar write:bar admin:all",
+                "expires_at": int(time.time()) + 500
+            },
+            {
+                "audience": "some_audience",
+                "access_token": "minimum_scope_token",
+                "scope": "read:foo write:foo read:bar write:bar",
+                "expires_at": int(time.time()) + 500
+            }
+        ]
+    }
+
+    client = ServerClient(
+        domain="auth0.local",
+        client_id="client_id",
+        client_secret="client_secret",
+        transaction_store=AsyncMock(),
+        state_store=mock_state_store,
+        secret="some-secret"
+    )
+
+    get_refresh_token_mock = mocker.patch.object(client, "get_token_by_refresh_token")
+
+    token = await client.get_access_token(
+        audience="some_audience",
+        scope="read:foo read:bar"
+    )
+
+    assert token == "minimum_scope_token"
+    get_refresh_token_mock.assert_not_awaited()
+
 @pytest.mark.asyncio
 async def test_get_access_token_for_connection_cached():
     mock_state_store = AsyncMock()