Allow reuse of stored tokens if tokenset scopes are a superset of requested scopes

Author: sam-muncke

src/auth0_server_python/auth_server/server_client.py

@@ -677,10 +677,9 @@ def _find_matching_token_set(
         match = None
         for token_set in token_sets:
             token_set_audience = token_set.get("audience")
-            matches_audience = token_set_audience == audience
             token_set_scopes = set(token_set.get("scope", "").split())
-            matches_scope = not scope or token_set_scopes == set(scope.split())
-            if matches_audience and matches_scope:
+            requested_scopes = set(scope.split()) if scope else set()
+            if token_set_audience == audience and token_set_scopes.issuperset(requested_scopes):
                 match = token_set
                 break
 

src/auth0_server_python/tests/test_server_client.py

@@ -657,6 +657,45 @@ async def test_get_access_token_from_store_with_multiple_audiences(mocker):
     assert token == "other_token_from_store"
     get_refresh_token_mock.assert_not_awaited()
 
+@pytest.mark.asyncio
+async def test_get_access_token_from_store_with_a_superset_of_requested_scopes(mocker):
+    mock_state_store = AsyncMock()
+    mock_state_store.get.return_value = {
+        "refresh_token": None,
+        "token_sets": [
+            {
+                "audience": "default",
+                "access_token": "token_from_store",
+                "expires_at": int(time.time()) + 500
+            },
+            {
+                "audience": "some_audience",
+                "access_token": "other_token_from_store",
+                "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 == "other_token_from_store"
+    get_refresh_token_mock.assert_not_awaited()
+
 @pytest.mark.asyncio
 async def test_get_access_token_for_connection_cached():
     mock_state_store = AsyncMock()