如何在 django 通道上使用 Knox 令牌身份验证来验证 websocket 连接?
How do you authenticate a websocket connection with Knox token authentication on django channels?
我知道您可以编写自定义身份验证中间件以在 Django 通道 2 中使用。
这适用于 Django 的内置令牌身份验证,但使用 django-rest-knox 令牌是另一回事。
Knox 以加密形式存储其令牌,因此它不像通过查找令牌从数据库中检索用户那样简单。
请帮忙。
想通了!
from knox.auth import TokenAuthentication
...
knoxAuth = TokenAuthentication();
user, auth_token = knoxAuth.authenticate_credentials(tokenString.encode(HTTP_HEADER_ENCODING))
scope['user'] = user
将以上代码与:https://gist.github.com/rluts/22e05ed8f53f97bdd02eafdf38f3d60a
整合
为了能够使用令牌验证来验证用户,您必须使用 cookie,headers 您可以使用 WS 发送是有限的,您还必须实现自己的“TokenAuthMiddleware”来处理 cookie .对于通道 2,您还必须正确处理对数据库的访问,下面是如何做到这一点:
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))
我知道您可以编写自定义身份验证中间件以在 Django 通道 2 中使用。 这适用于 Django 的内置令牌身份验证,但使用 django-rest-knox 令牌是另一回事。 Knox 以加密形式存储其令牌,因此它不像通过查找令牌从数据库中检索用户那样简单。 请帮忙。
想通了!
from knox.auth import TokenAuthentication
...
knoxAuth = TokenAuthentication();
user, auth_token = knoxAuth.authenticate_credentials(tokenString.encode(HTTP_HEADER_ENCODING))
scope['user'] = user
将以上代码与:https://gist.github.com/rluts/22e05ed8f53f97bdd02eafdf38f3d60a
整合为了能够使用令牌验证来验证用户,您必须使用 cookie,headers 您可以使用 WS 发送是有限的,您还必须实现自己的“TokenAuthMiddleware”来处理 cookie .对于通道 2,您还必须正确处理对数据库的访问,下面是如何做到这一点:
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))