Move function from models.IDToken to handler

Author: lullis

docs/settings.rst

@@ -409,6 +409,12 @@ When is set to ``False`` (default) the `OpenID Connect Backchannel Logout <https
 extension is not enabled. OpenID Connect Backchannel Logout enables the :term:`Authorization Server` (OpenID Provider) to submit a JWT token to an endpoint controlled by the :term:`Client` (Relying Party)
 indicating that a session from the :term:`Resource Owner` (End User) has ended.
 
+OIDC_BACKCHANNEL_LOGOUT_HANDLER
+~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
+Default: ``oauth2_provider.handlers.send_backchannel_logout_request``
+
+Upon logout, the :term:`Authorization Server` (OpenID Provider)  will look for all ID Tokens associated with the user on applications that support Backchannel Logout. For every id token that is found, the function defined here will be called. The default function can be used as-is, but if you need to override or customize it somehow (e.g, if you do not want to execute these requests on the same HTTP request-response from the user logout view), you can change this setting to any function that takes an IDToken as the first parameter.
+
 
 OIDC_RESPONSE_TYPES_SUPPORTED
 ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~

oauth2_provider/handlers.py

@@ -1,18 +1,84 @@
+import json
 import logging
+from datetime import timedelta
 
+import requests
 from django.contrib.auth.signals import user_logged_out
 from django.dispatch import receiver
+from django.utils import timezone
+from jwcrypto import jwt
 
+from .exceptions import BackchannelLogoutRequestError
+from .models import AbstractApplication, get_id_token_model
 from .settings import oauth2_settings
 
 
+IDToken = get_id_token_model()
+
 logger = logging.getLogger(__name__)
 
 
+def send_backchannel_logout_request(id_token, *args, **kwargs):
+    """
+    Send a logout token to the applications backchannel logout uri
+    """
+
+    ttl = kwargs.get("ttl") or timedelta(minutes=10)
+
+    try:
+        assert oauth2_settings.OIDC_BACKCHANNEL_LOGOUT_ENABLED, "Backchannel logout not enabled"
+        assert id_token.application.algorithm != AbstractApplication.NO_ALGORITHM, (
+            "Application must provide signing algorithm"
+        )
+        assert id_token.application.backchannel_logout_uri is not None, (
+            "URL for backchannel logout not provided by client"
+        )
+
+        issued_at = timezone.now()
+        expiration_date = issued_at + ttl
+
+        claims = {
+            "iss": oauth2_settings.OIDC_ISS_ENDPOINT,
+            "sub": str(id_token.user.id),
+            "aud": str(id_token.application.client_id),
+            "iat": int(issued_at.timestamp()),
+            "exp": int(expiration_date.timestamp()),
+            "jti": id_token.jti,
+            "events": {"http://schemas.openid.net/event/backchannel-logout": {}},
+        }
+
+        # Standard JWT header
+        header = {"typ": "logout+jwt", "alg": id_token.application.algorithm}
+
+        # RS256 consumers expect a kid in the header for verifying the token
+        if id_token.application.algorithm == AbstractApplication.RS256_ALGORITHM:
+            header["kid"] = id_token.application.jwk_key.thumbprint()
+
+        token = jwt.JWT(
+            header=json.dumps(header, default=str),
+            claims=json.dumps(claims, default=str),
+        )
+
+        token.make_signed_token(id_token.application.jwk_key)
+
+        headers = {"Content-Type": "application/x-www-form-urlencoded"}
+        data = {"logout_token": token.serialize()}
+        response = requests.post(id_token.application.backchannel_logout_uri, headers=headers, data=data)
+        response.raise_for_status()
+    except (AssertionError, requests.RequestException) as exc:
+        raise BackchannelLogoutRequestError(str(exc))
+
+
 @receiver(user_logged_out)
 def on_user_logged_out_maybe_send_backchannel_logout(sender, **kwargs):
     handler = oauth2_settings.OIDC_BACKCHANNEL_LOGOUT_HANDLER
     if not oauth2_settings.OIDC_BACKCHANNEL_LOGOUT_ENABLED or not callable(handler):
         return
 
-    handler(user=kwargs["user"])
+    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))

oauth2_provider/models.py

@@ -1,13 +1,11 @@
 import hashlib
-import json
 import logging
 import time
 import uuid
 from contextlib import suppress
 from datetime import timedelta
 from urllib.parse import parse_qsl, urlparse
 
-import requests
 from django.apps import apps
 from django.conf import settings
 from django.contrib.auth.hashers import identify_hasher, make_password
@@ -16,11 +14,10 @@
 from django.urls import reverse
 from django.utils import timezone
 from django.utils.translation import gettext_lazy as _
-from jwcrypto import jwk, jwt
+from jwcrypto import jwk
 from jwcrypto.common import base64url_encode
 from oauthlib.oauth2.rfc6749 import errors
 
-from .exceptions import BackchannelLogoutRequestError
 from .generators import generate_client_id, generate_client_secret
 from .scopes import get_scopes_backend
 from .settings import oauth2_settings
@@ -636,53 +633,6 @@ def revoke(self):
         """
         self.delete()
 
-    def send_backchannel_logout_request(self, ttl=timedelta(minutes=10)):
-        """
-        Send a logout token to the applications backchannel logout uri
-        """
-        try:
-            assert oauth2_settings.OIDC_BACKCHANNEL_LOGOUT_ENABLED, "Backchannel logout not enabled"
-            assert self.application.algorithm != AbstractApplication.NO_ALGORITHM, (
-                "Application must provide signing algorithm"
-            )
-            assert self.application.backchannel_logout_uri is not None, (
-                "URL for backchannel logout not provided by client"
-            )
-
-            issued_at = timezone.now()
-            expiration_date = issued_at + ttl
-
-            claims = {
-                "iss": oauth2_settings.OIDC_ISS_ENDPOINT,
-                "sub": str(self.user.id),
-                "aud": str(self.application.client_id),
-                "iat": int(issued_at.timestamp()),
-                "exp": int(expiration_date.timestamp()),
-                "jti": self.jti,
-                "events": {"http://schemas.openid.net/event/backchannel-logout": {}},
-            }
-
-            # Standard JWT header
-            header = {"typ": "logout+jwt", "alg": self.application.algorithm}
-
-            # RS256 consumers expect a kid in the header for verifying the token
-            if self.application.algorithm == AbstractApplication.RS256_ALGORITHM:
-                header["kid"] = self.application.jwk_key.thumbprint()
-
-            token = jwt.JWT(
-                header=json.dumps(header, default=str),
-                claims=json.dumps(claims, default=str),
-            )
-
-            token.make_signed_token(self.application.jwk_key)
-
-            headers = {"Content-Type": "application/x-www-form-urlencoded"}
-            data = {"logout_token": token.serialize()}
-            response = requests.post(self.application.backchannel_logout_uri, headers=headers, data=data)
-            response.raise_for_status()
-        except (AssertionError, requests.RequestException) as exc:
-            raise BackchannelLogoutRequestError(str(exc))
-
     @property
     def scopes(self):
         """
@@ -913,15 +863,3 @@ def is_origin_allowed(origin, allowed_origins):
             return True
 
     return False
-
-
-def send_backchannel_logout_requests(user):
-    """
-    Creates logout tokens for all id tokens associated with the user
-    """
-    id_tokens = IDToken.objects.filter(application__backchannel_logout_uri__isnull=False, user=user)
-    for id_token in id_tokens:
-        try:
-            id_token.send_backchannel_logout_request()
-        except BackchannelLogoutRequestError as exc:
-            logger.warn(str(exc))

oauth2_provider/settings.py

@@ -93,7 +93,7 @@
         "client_secret_basic",
     ],
     "OIDC_BACKCHANNEL_LOGOUT_ENABLED": False,
-    "OIDC_BACKCHANNEL_LOGOUT_HANDLER": "oauth2_provider.models.send_backchannel_logout_requests",
+    "OIDC_BACKCHANNEL_LOGOUT_HANDLER": "oauth2_provider.handlers.send_backchannel_logout_request",
     "OIDC_RP_INITIATED_LOGOUT_ENABLED": False,
     "OIDC_RP_INITIATED_LOGOUT_ALWAYS_PROMPT": True,
     "OIDC_RP_INITIATED_LOGOUT_STRICT_REDIRECT_URIS": False,

tests/test_backchannel_logout.py

@@ -8,7 +8,11 @@
 from django.urls import reverse
 
 from oauth2_provider.exceptions import BackchannelLogoutRequestError
-from oauth2_provider.models import get_application_model, get_id_token_model, send_backchannel_logout_requests
+from oauth2_provider.models import get_application_model, get_id_token_model
+from oauth2_provider.handlers import (
+    on_user_logged_out_maybe_send_backchannel_logout,
+    send_backchannel_logout_request,
+)
 from oauth2_provider.views import ApplicationRegistration
 
 from . import presets
@@ -42,12 +46,12 @@ def setUp(self):
         )
 
     def test_on_logout_handler_is_called_for_user(self):
-        with patch("oauth2_provider.models.send_backchannel_logout_requests") as backchannel_handler:
+        with patch("oauth2_provider.handlers.send_backchannel_logout_request") as backchannel_handler:
             self.client.login(username="app_user", password="654321")
             self.client.logout()
             backchannel_handler.assert_called_once()
-            args, kwargs = backchannel_handler.call_args
-            self.assertEqual(kwargs.get("user"), self.user)
+            _, kwargs = backchannel_handler.call_args
+            self.assertEqual(kwargs["id_token"], self.id_token)
 
     def test_logout_token_is_signed_for_user(self):
         with patch("requests.post") as mocked_post:
@@ -59,7 +63,7 @@ def test_raises_exception_on_bad_application(self):
         self.application.algorithm = Application.NO_ALGORITHM
         self.application.save()
         with self.assertRaises(BackchannelLogoutRequestError):
-            self.id_token.send_backchannel_logout_request()
+            send_backchannel_logout_request(self.id_token)
 
     def test_new_application_form_has_backchannel_logout_field(self):
         factory = RequestFactory()
@@ -71,6 +75,6 @@ def test_new_application_form_has_backchannel_logout_field(self):
         self.assertTrue("backchannel_logout_uri" in form.fields.keys())
 
     def test_logout_sender_does_not_crash_on_backchannel_error(self):
-        with patch.object(self.id_token, "send_backchannel_logout_request") as mock_func:
+        with patch("oauth2_provider.handlers.send_backchannel_logout_request") as mock_func:
             mock_func.side_effect = BackchannelLogoutRequestError("Bad Gateway")
-            send_backchannel_logout_requests(self.user)
+            on_user_logged_out_maybe_send_backchannel_logout(sender=User, user=self.user)