Django Channels middleware for routing public/protected WebSocket connections

Introduction

Django Channels is a Django extension that allows you to handle WebSockets, HTTP2, and other asynchronous protocols. It enables Django to support real-time, bi-directional communication channels between clients (such as web browsers) and the server. This is particularly useful for applications that require live updates, such as chat applications, real-time notifications, or collaborative editing tools.

Routing WebSocket Requests

To process all types of connections (HTTP, WebSocket) Django Channels hasProtocolTypeRouter class that choose the needed application depends on request type.

import os

from channels.auth import AuthMiddlewareStack
from channels.routing import ProtocolTypeRouter, URLRouter
from channels.security.websocket import AllowedHostsOriginValidator
from django.core.asgi import get_asgi_application
from django.urls import path

os.environ.setdefault("DJANGO_SETTINGS_MODULE", "mysite.settings")
# Initialize Django ASGI application early to ensure the AppRegistry
# is populated before importing code that may import ORM models.
django_asgi_app = get_asgi_application()

from chat.consumers import AdminChatConsumer, PublicChatConsumer

application = ProtocolTypeRouter({
    # Django's ASGI application to handle traditional HTTP requests
    "http": django_asgi_app,

    # WebSocket chat handler
    "websocket": AllowedHostsOriginValidator(
        AuthMiddlewareStack(
            URLRouter([
                path("chat/protected/", AdminChatConsumer.as_asgi()),
                path("chat/public", PublicChatConsumer.as_asgi()),
            ])
        )
    ),
})

As you can see from this example from the docs, all WebSocket routes are wrapped by AuthMiddlewareStack.

This middleware will try to authenticate and set the user into request scope. If there is no user it will be set user as Django AnonymousUser .

class AuthMiddleware(BaseMiddleware):
    """
    Middleware which populates scope["user"] from a Django session.
    Requires SessionMiddleware to function.
    """

    def populate_scope(self, scope):
        # Make sure we have a session
        if "session" not in scope:
            raise ValueError(
                "AuthMiddleware cannot find session in scope. "
                "SessionMiddleware must be above it."
            )
        # Add it to the scope if it's not there already
        if "user" not in scope:
            scope["user"] = UserLazyObject()
                                 . . .

And then in each consumer we have to check on user in scope that can be not convenient.

There is another way to split public/protected routes before the authorization. For this we will write a custom middleware that will process requests depends on them availability.

# middleware.py

class WebsocketPathMiddleware:

    def __init__(self, auth_middleware, app):
        self.app = app
        self.auth_middleware = auth_middleware(app)

    async def __call__(self, scope, receive, send):
        if self._is_route_public(scope.get("path", None)):
            return await self.app(scope, receive, send)
        return await self.auth_middleware(scope, receive, send)

    def _is_route_public(self, path: str) -> bool:
        path = path.lstrip("/")
        for route in public_websocket:
            match = route.pattern.match(path)
            if match:
                return True
        return False

This middleware class takes ASGI app and the authorization middleware to process protected requests.

# asgi.py

"""
ASGI entrypoint. Configures Django and then runs the application
defined in the ASGI_APPLICATION setting.
"""

import os

from channels.auth import AuthMiddlewareStack
from channels.routing import ProtocolTypeRouter, URLRouter
from channels.security.websocket import AllowedHostsOriginValidator
from django.core.asgi import get_asgi_application

os.environ.setdefault("DJANGO_SETTINGS_MODULE", "app.settings")
django_asgi_app = get_asgi_application()

from routing import public_websocket_routes, protected_websocket_routes
from middleware import WebsocketPathMiddleware


application = ProtocolTypeRouter(
    {
        "http": django_asgi_app,
        "websocket":
            AllowedHostsOriginValidator(
                WebsocketPathMiddleware(
                    AuthMiddlewareStack,
                    URLRouter(
                        [*public_websocket_routes, *protected_websocket_routes]
                    )
                )
            )
    }
)

# routing.py

from django.urls import path
from chat.consumers import AdminChatConsumer, PublicChatConsumer


public_websocket_routes = [
    path("chat/public", PublicChatConsumer.as_asgi())
]

protected_websocket_routes = [
    path("chat/protected/", AdminChatConsumer.as_asgi())
]

In Plain English 🚀

Thank you for being a part of the In Plain English community! Before you go:

• Be sure to clap and follow the writer ️👏️️
• Follow us: X | LinkedIn | YouTube | Discord | Newsletter
• Visit our other platforms: Stackademic | CoFeed | Venture | Cubed
• More content at PlainEnglish.io