BaseAuthTokenMiddleware

[YegorDB]

What do you think about add BaseAuthTokenMiddleware?

Django REST framework auth token middleware could look like this

import re

from django.apps import apps

from channels.db import database_sync_to_async
from channels.auth import BaseAuthTokenMiddleware


class DRFAuthTokenMiddleware(BaseAuthTokenMiddleware):

    keyword = 'Token'

    def parse_token_key(self, scope):
        headers = dict(scope['headers'])
        key = headers.get(b'authorization', b'').decode()
        matched = re.fullmatch(rf'{self.keyword} ([0-9a-f]{{40}})', key)
        if not matched:
            return None
        return matched.group(1)

    @database_sync_to_async
    def get_user_instance(self, token_key):
        Token = apps.get_model('authtoken', 'Token')
        try:
            token = Token.objects.select_related('user').get(key=token_key)
        except Token.DoesNotExist:
            return None
        return token.user

[YegorDB]

Since there is no reaction here, I made additional app to work with auth token middlewares.

[sergioisidoro]

@YegorDB have you considered making this PR at https://github.com/LostMoa/djangochannelsrestframework ?

[YegorDB]

Cool project, but I haven't seen it before. I think, since additional app was created, there is no need to make this PR at project you talk about. Furthermore in future there will be more middlewares (not only for Django REST framework) in additional app.

[johnthagen]

channels-auth-token-middlewares provides a QueryStringSimpleJWTAuthTokenMiddleware to support token authentication out of the box when using Simple JWT with Django REST Framework.

Update INSTALLED_APPS:

INSTALLED_APPS = [
    # base django apps (django.contrib.auth is required)
    # other apps this one depends on (like rest_framework if it's necessary)
    'channels_auth_token_middlewares',
    # custom apps
]

Insert QueryStringSimpleJWTAuthTokenMiddleware into your ASGI application stack:

application = ProtocolTypeRouter(
    {
        "http": django_asgi_app,
        "websocket": AllowedHostsOriginValidator(
            QueryStringSimpleJWTAuthTokenMiddleware(
                URLRouter(...),
            ),
        ),
    }
)

Clients pass their JWT token into the token query parameter:

from websocket import create_connection

token = "EXAMPLE_TOKEN"
ws = create_connection(f"ws://127.0.0.1/ws/?token={token}")

The authenticated User (or AnonymousUser if the JWT is invalid) will be populated into the "user" key of the scope passed to the Consumer

class MyAsyncCommunicator(AsyncWebsocketConsumer):
    async def connect(self) -> None:
        user = self.scope["user"]
        # Validate user before accepting the Websocket Connection
        # For example:
        if not user.is_authenticated or user.is_anonymous:
            # Handle unauthorized.