Skip to content

Instantly share code, notes, and snippets.

@Coronon
Created December 28, 2020 00:52
Show Gist options
  • Star 0 You must be signed in to star a gist
  • Fork 0 You must be signed in to fork a gist
  • Save Coronon/e79fb25ae9ee83fc691dd5b4cdac2231 to your computer and use it in GitHub Desktop.
Save Coronon/e79fb25ae9ee83fc691dd5b4cdac2231 to your computer and use it in GitHub Desktop.
Django Channels Websocket logout handler (Handles all logout throughout the app)
import functools
from asgiref.sync import async_to_sync
from django.contrib.auth.signals import user_logged_out
from channels.layers import get_channel_layer
def __ws_login_required_signal_handler(sender, request, user, **kwargs):
print(request.session.session_key)
channel_layer = get_channel_layer()
async_to_sync(channel_layer.group_send)(f"SESSION-{request.session.session_key}", {"type": "close", "code": None})
user_logged_out.connect(__ws_login_required_signal_handler, dispatch_uid="__ws_login_required_signal_handler")
def ws_login_required(cls):
"""
Decorator for ws-channels that checks that the user is logged in,
closing the connection if necessary (Only works on sync ws-consumers).
"""
def ws_login_required_connect(func):
"""
Decorator for ws-channels that checks that the user is logged in,
closing the connection if necessary (Only works on 'connect').
"""
@functools.wraps(func)
def wrapper_ws_login_required_connect(self, *args, **kwargs):
if self.scope["user"].is_authenticated:
async_to_sync(self.channel_layer.group_add)(f"SESSION-{self.scope['session'].session_key}", self.channel_name)
return func(self, *args, **kwargs)
else:
self.close()
return wrapper_ws_login_required_connect
def ws_login_required_close(func):
"""
Decorator for ws-channels that checks that the user is logged in,
closing the connection if necessary (Only works on 'close').
"""
@functools.wraps(func)
def wrapper_ws_login_required_close(self, code=None, *args, **kwargs):
if isinstance(code, dict):
code = code["code"] if "code" in code else None
return func(self, code, *args, **kwargs)
return wrapper_ws_login_required_close
cls.connect = ws_login_required_connect(cls.connect)
cls.close = ws_login_required_close(cls.close)
return cls
@Coronon
Copy link
Author

Coronon commented Dec 28, 2020

Just decorate your SYNC websocket class with @ws_login_required and all logouts of the session used to authenticate the websocket connection to a consumer decorated, will be logged out. (Multi connection support) (Differentiates between devices)

@Coronon
Copy link
Author

Coronon commented Dec 28, 2020

A more robust solution with multiple files and safer signal subscribing can be found here: django/channels#1324 (comment)

Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment