CODEWITHSUNDEEP
authenticationwebsocketdjango-rest-frameworkdjango-channelsdjango-rest-knox
Brian Min
Brian Min

Reputation: 297

How do you authenticate a websocket connection with Knox token authentication on django channels?

I understand you can write custom authentication middleware to use in django channels 2. This works fine with Django's built-in token authentication but using django-rest-knox tokens is a different story. Knox stores its tokens in an encrypted form so it is not as easy as simply retrieving the user from the database by looking up the token. Please help.

Upvotes: 3

Views: 1105

Answers (2)

user7870824
user7870824

Reputation:

In order to be able to authenticate a user using token authentication, you must use cookies, The headers you can send using WS are limited, you must also implement your own "TokenAuthMiddleware" to handle the cookie. for channels 2, you also have to handle access to the database correctly, below is how to do that:

from    channels.auth                   import  AuthMiddlewareStack
from    channels.db                     import  database_sync_to_async
from    knox.auth                       import  TokenAuthentication
from    django.contrib.auth.models      import  AnonymousUser
from    django.db                       import  close_old_connections
from    rest_framework.exceptions       import  AuthenticationFailed

import  re

class TokenAuthMiddlewareInstance   :
    def __init__            (
        #
                self        ,
                scope       ,
                middleware  ,
            ):
        self.middleware = middleware
        self.scope      = dict(scope)
        self.inner      = self.middleware.inner
    async def __call__      (
        #
                self    ,
                receive ,
                send    ,
            ):
        self.scope['user']  = AnonymousUser()
        cookie              = dict(self.scope.get('headers',{})).get(b'cookie')

        if      cookie  :
            token   = re.findall(r'X-Authorization=(\w*)', cookie.decode('ascii'))
            if      len(token)  :
                self.scope['user']   = await self._g_user(token)

        inner = self.inner(self.scope)
        return await inner(receive, send)
    
    @database_sync_to_async
    def _g_user (
        #
                self    ,
                token   ,
            ):
        try                                 :
            token_key       = token[0]
            user, token     = TokenAuthentication().authenticate_credentials(token_key.encode('ascii'))
            close_old_connections()
            return user
        except AuthenticationFailed as e    :
            return AnonymousUser()
class TokenAuthMiddleware           :
    def __init__    (
        #
                self    ,
                inner   ,
            ):
        self.inner = inner

    def __call__    (
        #
                self    ,
                scope   ,
            ):
        return TokenAuthMiddlewareInstance(scope, self)
TokenAuthMiddlewareStack            = lambda inner: TokenAuthMiddleware(AuthMiddlewareStack(inner))

Upvotes: 1

Brian Min
Brian Min

Reputation: 297

Figured it out!

from knox.auth import TokenAuthentication
...
knoxAuth = TokenAuthentication();
user, auth_token = knoxAuth.authenticate_credentials(tokenString.encode(HTTP_HEADER_ENCODING))
scope['user'] = user

Integrate the above code with: https://gist.github.com/rluts/22e05ed8f53f97bdd02eafdf38f3d60a

Upvotes: 5

Related Questions