Send only one logout request per application, per user

Author: lullis

oauth2_provider/handlers.py

@@ -9,11 +9,12 @@
 from jwcrypto import jwt
 
 from .exceptions import BackchannelLogoutRequestError
-from .models import AbstractApplication, get_id_token_model
+from .models import AbstractApplication, get_id_token_model, get_refresh_token_model
 from .settings import oauth2_settings
 
 
 IDToken = get_id_token_model()
+RefreshToken = get_refresh_token_model()
 
 logger = logging.getLogger(__name__)
 
@@ -76,9 +77,22 @@ def on_user_logged_out_maybe_send_backchannel_logout(sender, **kwargs):
         return
 
     user = kwargs["user"]
-    id_tokens = IDToken.objects.filter(application__backchannel_logout_uri__isnull=False, user=user)
-    for id_token in id_tokens:
-        try:
-            handler(id_token=id_token)
-        except BackchannelLogoutRequestError as exc:
-            logger.warn(str(exc))
+    # Get refresh tokens for user where scope doesn't contain offline_access
+    refresh_tokens = (
+        RefreshToken.objects.filter(user=user, access_token__scope__isnull=False)
+        .exclude(access_token__scope__icontains="offline_access")
+        .select_related("application", "access_token")
+    )
+
+    # Group by application and send one request per application
+    applications_to_logout = set()
+    for rt in refresh_tokens:
+        if rt.application.backchannel_logout_uri and rt.application not in applications_to_logout:
+            applications_to_logout.add(rt.application)
+            # Find an ID token for this user/application to use for logout
+            id_token = IDToken.objects.filter(application=rt.application, user=user).first()
+            if id_token:
+                try:
+                    handler(id_token=id_token)
+                except BackchannelLogoutRequestError as exc:
+                    logger.warn(str(exc))

tests/test_backchannel_logout.py

@@ -8,7 +8,12 @@
 from django.urls import reverse
 
 from oauth2_provider.exceptions import BackchannelLogoutRequestError
-from oauth2_provider.models import get_application_model, get_id_token_model
+from oauth2_provider.models import (
+    get_application_model,
+    get_id_token_model,
+    get_access_token_model,
+    get_refresh_token_model,
+)
 from oauth2_provider.handlers import (
     on_user_logged_out_maybe_send_backchannel_logout,
     send_backchannel_logout_request,
@@ -21,6 +26,8 @@
 
 Application = get_application_model()
 IDToken = get_id_token_model()
+AccessToken = get_access_token_model()
+RefreshToken = get_refresh_token_model()
 User = get_user_model()
 
 
@@ -45,6 +52,20 @@ def setUp(self):
             application=self.application, user=self.user, expires=expiration_date
         )
 
+        self.access_token = AccessToken.objects.create(
+            user=self.user,
+            application=self.application,
+            token="test_access_token",
+            expires=now + datetime.timedelta(hours=1),
+            scope="read write",  # No offline_access scope
+        )
+        self.refresh_token = RefreshToken.objects.create(
+            user=self.user,
+            application=self.application,
+            token="test_refresh_token",
+            access_token=self.access_token,
+        )
+
     def test_on_logout_handler_is_called_for_user(self):
         with patch("oauth2_provider.handlers.send_backchannel_logout_request") as backchannel_handler:
             self.client.login(username="app_user", password="654321")
@@ -78,3 +99,84 @@ def test_logout_sender_does_not_crash_on_backchannel_error(self):
         with patch("oauth2_provider.handlers.send_backchannel_logout_request") as mock_func:
             mock_func.side_effect = BackchannelLogoutRequestError("Bad Gateway")
             on_user_logged_out_maybe_send_backchannel_logout(sender=User, user=self.user)
+
+    def test_no_logout_sent_when_only_offline_access_refresh_tokens(self):
+        # Add offline_access scope
+        self.access_token.scope = "read write offline_access"
+        self.access_token.save()
+
+        with patch("oauth2_provider.handlers.send_backchannel_logout_request") as backchannel_handler:
+            on_user_logged_out_maybe_send_backchannel_logout(sender=User, user=self.user)
+            backchannel_handler.assert_not_called()
+
+    def test_logout_sent_when_refresh_token_without_offline_access(self):
+        with patch("oauth2_provider.handlers.send_backchannel_logout_request") as backchannel_handler:
+            on_user_logged_out_maybe_send_backchannel_logout(sender=User, user=self.user)
+            backchannel_handler.assert_called_once()
+            _, kwargs = backchannel_handler.call_args
+            self.assertEqual(kwargs["id_token"], self.id_token)
+
+    def test_only_one_logout_per_application_with_multiple_refresh_tokens(self):
+        # Create another refresh token for the same application
+        now = timezone.now()
+        another_access_token = AccessToken.objects.create(
+            user=self.user,
+            application=self.application,
+            token="test_access_token_2",
+            expires=now + datetime.timedelta(hours=1),
+            scope="read write",
+        )
+        RefreshToken.objects.create(
+            user=self.user,
+            application=self.application,
+            token="test_refresh_token_2",
+            access_token=another_access_token,
+        )
+
+        # Should still be called only once despite having 2 refresh tokens
+        with patch("oauth2_provider.handlers.send_backchannel_logout_request") as backchannel_handler:
+            on_user_logged_out_maybe_send_backchannel_logout(sender=User, user=self.user)
+            backchannel_handler.assert_called_once()
+
+    def test_logout_sent_for_multiple_applications(self):
+        # Create another application with backchannel logout URI
+        another_app = Application.objects.create(
+            name="test_app_2",
+            user=self.developer,
+            client_type=Application.CLIENT_PUBLIC,
+            authorization_grant_type=Application.GRANT_CLIENT_CREDENTIALS,
+            algorithm=Application.RS256_ALGORITHM,
+            client_secret="another_secret",
+            backchannel_logout_uri="http://rp2.example.com/logout",
+        )
+
+        # Create ID token for the second application
+        another_id_token = IDToken.objects.create(
+            application=another_app, user=self.user, expires=timezone.now() + datetime.timedelta(minutes=180)
+        )
+
+        # Create access token and refresh token for the second application
+        now = timezone.now()
+        another_access_token = AccessToken.objects.create(
+            user=self.user,
+            application=another_app,
+            token="test_access_token_app2",
+            expires=now + datetime.timedelta(hours=1),
+            scope="read write",
+        )
+        RefreshToken.objects.create(
+            user=self.user,
+            application=another_app,
+            token="test_refresh_token_app2",
+            access_token=another_access_token,
+        )
+
+        # Should be called twice - once for each application - and both ID tokens were used
+        with patch("oauth2_provider.handlers.send_backchannel_logout_request") as backchannel_handler:
+            on_user_logged_out_maybe_send_backchannel_logout(sender=User, user=self.user)
+            self.assertEqual(backchannel_handler.call_count, 2)
+
+            call_args_list = backchannel_handler.call_args_list
+            id_tokens_called = [call.kwargs["id_token"] for call in call_args_list]
+            self.assertIn(self.id_token, id_tokens_called)
+            self.assertIn(another_id_token, id_tokens_called)