fix: scoped logout, backchannel logout

Author: kishore7snehil

src/auth0_server_python/auth_server/server_client.py

@@ -768,13 +768,22 @@ async def logout(
     ) -> str:
         options = options or LogoutOptions()
 
-        # Delete the session from the state store
-        await self._state_store.delete(self._state_identifier, store_options)
+        if not self._domain_resolver:
+            await self._state_store.delete(self._state_identifier, store_options)
+            domain = self._domain
+        else:
+            # Resolver mode: delete session if domains match
+            domain = await self._resolve_current_domain(store_options)
+            state_data = await self._state_store.get(self._state_identifier, store_options)
 
-        # Resolve domain dynamically for MCD support
-        domain = await self._resolve_current_domain(store_options)
+            if state_data:
+                if hasattr(state_data, "dict") and callable(state_data.dict):
+                    state_data = state_data.dict()
+                session_domain = state_data.get("domain")
+                if session_domain and self._normalize_issuer(session_domain) == self._normalize_issuer(domain):
+                    await self._state_store.delete(self._state_identifier, store_options)
 
-        # Use the URL helper to create the logout URL.
+        # Return logout URL for the current resolved domain
         logout_url = URL.create_logout_url(
             domain, self._client_id, options.return_to)
 
@@ -828,7 +837,7 @@ async def handle_backchannel_logout(
             jwks = await self._get_jwks_cached(domain)
 
             try:
-                claims = await self._verify_and_decode_jwt(logout_token, jwks)
+                claims = await self._verify_and_decode_jwt(logout_token, jwks, audience=self._client_id)
 
                 # Normalized issuer validation
                 token_issuer = claims.get("iss", "")
@@ -861,7 +870,12 @@ async def handle_backchannel_logout(
                 sid=claims.get("sid")
             )
 
-            await self._state_store.delete_by_logout_token(logout_claims.dict(), store_options)
+            # In resolver mode, include iss for issuer-scoped deletion
+            claims_dict = logout_claims.dict()
+            if self._domain_resolver:
+                claims_dict["iss"] = claims.get("iss")
+
+            await self._state_store.delete_by_logout_token(claims_dict, store_options)
 
         except (jwt.PyJWTError, ValidationError) as e:
             raise BackchannelLogoutError(
@@ -1473,7 +1487,9 @@ async def start_link_user(
 
         # In resolver mode, reject sessions without domain or with mismatched domain
         if self._domain_resolver:
-            session_domain = state_data.get('domain') if isinstance(state_data, dict) else getattr(state_data, 'domain', None)
+            if hasattr(state_data, "dict") and callable(state_data.dict):
+                state_data = state_data.dict()
+            session_domain = state_data.get('domain')
             if not session_domain:
                 raise StartLinkUserError(
                     "Session is missing domain. User needs to re-authenticate."
@@ -1568,7 +1584,9 @@ async def start_unlink_user(
 
         # In resolver mode, reject sessions without domain or with mismatched domain
         if self._domain_resolver:
-            session_domain = state_data.get('domain') if isinstance(state_data, dict) else getattr(state_data, 'domain', None)
+            if hasattr(state_data, "dict") and callable(state_data.dict):
+                state_data = state_data.dict()
+            session_domain = state_data.get('domain')
             if not session_domain:
                 raise StartLinkUserError(
                     "Session is missing domain. User needs to re-authenticate."

src/auth0_server_python/store/abstract.py

@@ -96,7 +96,8 @@ async def delete_by_logout_token(self, claims: dict[str, Any], options: Optional
         Delete sessions based on logout token claims.
 
         Args:
-            claims: Claims from the logout token
+            claims: Claims from the logout token (sub, sid, and optionally iss
+                    in MCD mode for issuer-scoped deletion)
             options: Additional operation-specific options
 
         Note:

src/auth0_server_python/tests/test_server_client.py

@@ -1445,14 +1445,20 @@ async def test_handle_backchannel_logout_ok(mocker):
     mock_signing_key = mocker.MagicMock()
     mock_signing_key.key = "mock_pem_key"
     mocker.patch("jwt.PyJWK.from_dict", return_value=mock_signing_key)
-    mocker.patch("jwt.decode", return_value={
+    mock_jwt_decode = mocker.patch("jwt.decode", return_value={
         "events": {"http://schemas.openid.net/event/backchannel-logout": {}},
         "iss": "https://auth0.local",
         "sub": "user_sub",
         "sid": "session_id_123"
     })
 
     await client.handle_backchannel_logout("some_logout_token")
+
+    # Verify audience is passed to jwt.decode
+    call_kwargs = mock_jwt_decode.call_args[1]
+    assert call_kwargs["audience"] == "client_id"
+
+    # In static mode, iss should NOT be included
     mock_state_store.delete_by_logout_token.assert_awaited_once_with(
         {"sub": "user_sub", "sid": "session_id_123"},
         None
@@ -1498,7 +1504,7 @@ async def domain_resolver(context):
         return_value={"keys": [{"kty": "RSA", "kid": "test-key"}]}
     )
 
-    mocker.patch.object(client, "_verify_and_decode_jwt", return_value={
+    mock_verify = mocker.patch.object(client, "_verify_and_decode_jwt", return_value={
         "iss": "https://tenant1.auth0.com/",
         "events": {"http://schemas.openid.net/event/backchannel-logout": {}},
         "sub": "user123",
@@ -1507,8 +1513,14 @@ async def domain_resolver(context):
 
     await client.handle_backchannel_logout("some_logout_token")
 
+    # Verify audience is passed to JWT verification
+    mock_verify.assert_awaited_once()
+    call_kwargs = mock_verify.call_args[1]
+    assert call_kwargs["audience"] == "test_client"
+
+    # In resolver mode, iss should be included for issuer-scoped deletion
     mock_state_store.delete_by_logout_token.assert_awaited_once_with(
-        {"sub": "user123", "sid": "session123"},
+        {"sub": "user123", "sid": "session123", "iss": "https://tenant1.auth0.com/"},
         None
     )
 
@@ -4405,6 +4417,7 @@ async def domain_resolver(context):
         return current_domain
 
     mock_state_store = AsyncMock()
+    mock_state_store.get.return_value = {"domain": current_domain, "user": {"sub": "user1"}}
 
     client = ServerClient(
         domain=domain_resolver,
@@ -4420,6 +4433,8 @@ async def domain_resolver(context):
     # Verify logout URL uses current domain
     assert current_domain in logout_url
     assert logout_url.startswith(f"https://{current_domain}")
+    # Verify session was deleted (domains match)
+    mock_state_store.delete.assert_called_once()
 
 
 @pytest.mark.asyncio