Authentication
Chanx provides a robust authentication system for WebSockets that seamlessly integrates with Django REST Framework's authentication and permission classes. This allows you to secure your WebSocket endpoints using the same mechanisms you already use for your REST API.
How Authentication Works
When a WebSocket connection is established:
The connection scope is converted to a Django request object
DRF authentication classes process the request
Permission classes verify access rights
If authentication succeeds, the connection is accepted
If authentication fails, the connection is closed with an error message
Chanx supports authentication via cookies or query parameters that are passed during the initial WebSocket handshake. Since browsers don't allow custom headers in WebSocket connections, cookie-based authentication is the recommended approach for browser clients.
Configuration
To configure authentication for a WebSocket consumer, set the authentication_classes and permission_classes attributes:
from rest_framework.authentication import SessionAuthentication
from rest_framework.permissions import IsAuthenticated
from chanx.generic.websocket import AsyncJsonWebsocketConsumer
from myapp.messages import MyIncomingMessage
class SecureConsumer(AsyncJsonWebsocketConsumer[MyIncomingMessage]):
authentication_classes = [SessionAuthentication] # Cookie-based authentication
permission_classes = [IsAuthenticated]
async def receive_message(self, message: MyIncomingMessage, **kwargs: Any) -> None:
# Only authenticated users reach this point
match message:
case PingMessage():
await self.send_message(PongMessage())
case _:
# Handle other message types
pass
Client-Side Authentication Best Practices
For browser-based WebSocket clients, cookie authentication is the most straightforward approach:
Session Authentication: Have the user log in through your regular Django views or REST API
JWT in HTTP-only Cookie: For token-based auth, store the JWT in an HTTP-only cookie
Query Parameters: For simple testing or non-browser clients, query parameters can be used
Example using HTTP-only cookie (recommended for browsers):
// JavaScript WebSocket client with cookie auth
// (Cookie is automatically included by the browser)
const socket = new WebSocket('ws://example.com/ws/endpoint/');
For non-browser clients or testing, query parameters can be used:
// Using query parameter for token
const socket = new WebSocket('ws://example.com/ws/endpoint/?token=your-auth-token');
Object-Level Permissions
Chanx supports object-level permissions just like DRF. To use them:
Set a
queryseton your consumerUse permission classes with
has_object_permissionSpecify the model type as the fourth generic parameter
from rest_framework.permissions import BasePermission
from myapp.models import Room
class RoomAccessPermission(BasePermission):
def has_object_permission(self, request, view, obj):
# Check if user is a member of this room
return request.user in obj.members.all()
class RoomConsumer(AsyncJsonWebsocketConsumer[ChatIncomingMessage, None, Room]):
authentication_classes = [SessionAuthentication]
permission_classes = [IsAuthenticated, RoomAccessPermission]
queryset = Room.objects.all()
async def build_groups(self) -> list[str]:
# self.obj now contains the Room instance
# and is properly typed as Room
assert self.obj
return [f"room_{self.obj.id}"]
With this setup, Chanx will:
Extract the lookup parameter from the URL
Retrieve the object from the queryset
Check object-level permissions
Make the object available as
self.objin the consumer
Authentication Messages
By default, Chanx sends an authentication status message when a client connects. You can control this with the send_authentication_message setting:
class MyConsumer(AsyncJsonWebsocketConsumer[MyIncomingMessage]):
send_authentication_message = True # Default is True
The authentication message looks like:
{
"action": "authentication",
"payload": {
"status_code": 200,
"status_text": "OK",
"data": {
"detail": "OK"
}
}
}
Or on failure:
{
"action": "authentication",
"payload": {
"status_code": 403,
"status_text": "Forbidden",
"data": {
"detail": "Authentication credentials were not provided."
}
}
}
Custom Authentication
For more advanced authentication needs, you can create a custom authenticator by extending the ChanxWebsocketAuthenticator class:
from chanx.generic.authenticator import ChanxWebsocketAuthenticator, AuthenticationResult
class MyAuthenticator(ChanxWebsocketAuthenticator):
async def authenticate(self, scope):
# First perform the standard authentication
auth_result = await super().authenticate(scope)
# Add additional validation or processing
if auth_result.is_authenticated:
# Example: Check if user is active in the current module
user = auth_result.user
if not await is_user_active_in_module(user):
# Override authentication result
return AuthenticationResult(
is_authenticated=False,
status_code=403,
status_text="Forbidden",
data={"detail": "User is not active in this module"},
user=user,
obj=None,
)
return auth_result
class MyConsumer(AsyncJsonWebsocketConsumer[MyIncomingMessage]):
authenticator_class = MyAuthenticator
Best Practices
Use HTTP-only cookies for browser-based clients to prevent XSS vulnerabilities
Keep authentication consistent between your REST API and WebSockets
Test authentication thoroughly, including failure scenarios
Use object-level permissions when endpoints deal with specific resources
Avoid storing sensitive tokens in JavaScript variables or localStorage
Set appropriate cookie security flags (Secure, SameSite) in production
Implement periodic token validation for long-lived connections
Use generic type parameters for better type checking of models
Next Steps
Consumers - Learn about configuring consumers
Testing - More on testing WebSocket endpoints
Chat Application Example - See authentication in a complete example