authentik.providers.oauth2.tasks

 1"""OAuth2 Provider Tasks"""
 2
 3from django.utils.translation import gettext_lazy as _
 4from dramatiq.actor import actor
 5from structlog.stdlib import get_logger
 6
 7from authentik.lib.utils.http import get_http_session
 8from authentik.providers.oauth2.models import OAuth2LogoutMethod, OAuth2Provider
 9from authentik.providers.oauth2.utils import create_logout_token
10from authentik.tasks.middleware import CurrentTask
11
12LOGGER = get_logger()
13
14
15@actor(description=_("Send a back-channel logout request to the registered client"))
16def send_backchannel_logout_request(
17    provider_pk: int,
18    iss: str,
19    sub: str | None = None,
20    session_key: str | None = None,
21) -> bool:
22    """Send a back-channel logout request to the registered client
23
24    Args:
25        provider_pk: The OAuth2 provider's primary key
26        iss: The issuer URL for the logout token
27        sub: The subject identifier to include in the logout token
28        session_key: The authentik session key to hash and include in the logout token
29
30    Returns:
31        bool: True if the request was sent successfully, False otherwise
32    """
33    self = CurrentTask.get_task()
34    LOGGER.debug("Sending back-channel logout request", provider_pk=provider_pk, sub=sub)
35
36    provider = OAuth2Provider.objects.filter(pk=provider_pk).first()
37    if provider is None:
38        return
39
40    # Generate the logout token
41    logout_token = create_logout_token(provider, iss, sub, session_key)
42
43    if provider.logout_method != OAuth2LogoutMethod.BACKCHANNEL:
44        self.info("Provider not configured for back-channel logout")
45        return
46
47    if not provider.logout_uri:
48        self.info("No logout URI configured for provider")
49        return
50
51    # Send the back-channel logout request
52    response = get_http_session().post(
53        provider.logout_uri,
54        data={"logout_token": logout_token},
55        headers={"Content-Type": "application/x-www-form-urlencoded"},
56        allow_redirects=True,
57    )
58    response.raise_for_status()
59
60    self.info("Back-channel logout successful", sub=sub)
61    return True
62
63
64@actor(description=_("Handle backchannel logout notifications dispatched via signal"))
65def backchannel_logout_notification_dispatch(revocations: list, **kwargs):
66    """Handle backchannel logout notifications dispatched via signal"""
67    for revocation in revocations:
68        provider_pk, iss, sub, session_key = revocation
69        provider = OAuth2Provider.objects.filter(pk=provider_pk).first()
70        send_backchannel_logout_request.send_with_options(
71            args=(provider_pk, iss, sub, session_key),
72            rel_obj=provider,
73        )