Custom AuthMiddlewareStack to get users from JWT token for Django Channels.
Simply,
$ pip install django-channels-jwt-auth-middleware
And that it.
All you have to do is wrap your URLRouter.
from django_channels_jwt_auth_middleware.auth import JWTAuthMiddlewareStack
application = ProtocolTypeRouter({
"http": get_asgi_application(),
"websocket": JWTAuthMiddlewareStack(
URLRouter(
websocket_urlpatterns,
)
),
})
Then, add your JWT token to your websocket path.
/path/to/consumer/?token=<your JWT token>
JWTAuthMiddlewareStack
will get the user_id from your JWT and get the user based on it. If it finds the user then it will attach that user to self.scope['user']
. If it does not find a user the Middleware will sttach an AnonymousUser to the scope.
AuthMiddlewareStack
is already in JWTAuthMiddlewareStack
. If you want to change this, simply do;
from django_channels_jwt_auth_middleware.auth import JWTAuthMiddleware
from . import YourCustomMiddlewareStack
def CustomJWTAuthMiddlewareStack(app):
return JWTAuthMiddleware(YourCustomMiddlewareStack(app))
Simple as that :)
Testing is done by two methods.
- automated testing using django's test system.
- manual testing by hand.
I have tried testing the middleware through ChannelsLiveServerTestCase
, but currently this does not run due to pickle error in multiprocessing package from python.
For this reason no-token test cases were tested using django's test system using a test project, and test cases with jwt token is tested by hand.
For hand testing chrome extension called Browser WebSocket Client
was used. Blow is a simple evidence from the hand testing.
If you go to testproject/pages/tests.py
you will see the consumer that is used for testing.
class TestConsumer(WebsocketConsumer):
def connect(self):
self.accept()
# def disconnect(self, code):
# self.disconnect()
def receive(self, text_data, bytes_data=None):
context = {
'message': 'testing consumer',
}
user = self.scope['user']
if (user_id := user.id):
context.update({'user': str(user_id)})
else:
context.update({'user': None})
self.send(text_data=json.dumps(context))
simply, if a data is passed in to the websocket it will return a user id of None for AnonymousUser.