Simple Jwt Token AuthMiddleware for channels 2.0

_JwtTokenAuthMiddleware.md

Incase you are using djangorestframework-simplejwt for authentication and Would like to use channels==2.0.

On web sending token over headers is not possible if using js WebSocket available API. To go around this you can send your token on the url as indicated below wws://example.com?token=[a_valid_token]

The middleware above can be added on routing.py to supply a valid user to the consumer.

NB :: Securitywise this is not advisable to do.

auth.py

class JwtTokenAuthMiddleware(BaseMiddleware):
    """
    JWT token authorization middleware for Django Channels 2
    """

    def get_validated_token(self, raw_token):
        """
        Validates an encoded JSON web token and returns a validated token
        wrapper object.
        """
        messages = []
        for AuthToken in api_settings.AUTH_TOKEN_CLASSES:
            try:
                return AuthToken(raw_token)
            except TokenError as e:
                messages.append({'token_class': AuthToken.__name__,
                                 'token_type': AuthToken.token_type,
                                 'message': e.args[0]})

        raise InvalidToken({
            'detail': _('Given token not valid for any token type'),
            'messages': messages,
        })

    def get_user(self, validated_token):
        """
        Attempts to find and return a user using the given validated token.
        """
        try:
            user_id = validated_token[api_settings.USER_ID_CLAIM]
        except KeyError:
            raise InvalidToken(_('Token contained no recognizable user identification'))

        try:
            user = User.objects.get(**{api_settings.USER_ID_FIELD: user_id})
        except User.DoesNotExist:
            raise AuthenticationFailed(_('User not found'), code='user_not_found')

        if not user.is_active:
            raise AuthenticationFailed(_('User is inactive'), code='user_inactive')

        return user

    def __init__(self, inner):
        self.inner = inner

    def __call__(self, scope):
        try:
            raw_token = scope['query_string'].decode().split('=')[1]
            validated_token = self.get_validated_token(raw_token)
            user = self.get_user(validated_token=validated_token)
            scope['user'] = user
        except:
            pass
        return self.inner(scope)


JwtTokenAuthMiddlewareStack = lambda inner: JwtTokenAuthMiddleware(AuthMiddlewareStack(inner))