channels 使用 djangorestframework-simplejwt 的token 验证用户有效性

2022-08-13 14:05:53

前言

channels 使用 djangorestframework-simplejwt 的token 验证用户有效性


一、安装好channels库后,在asgi设置好websocket路由

官方推荐的写法

# mysite/asgi.pyimport osfrom channels.authimport AuthMiddlewareStackfrom channels.routingimport ProtocolTypeRouter, URLRouterfrom django.core.asgiimport get_asgi_applicationimport chat.routing

os.environ.setdefault("DJANGO_SETTINGS_MODULE","mysite.settings")

application= ProtocolTypeRouter({"http": get_asgi_application(),"websocket": AuthMiddlewareStack(
        URLRouter(
            chat.routing.websocket_urlpatterns)),})

但是 AuthMiddlewareStack 并不能帮我们解析到用户实例,包括我们要自定义的方法,很不方便,而且我用的 djangorestframework-simplejwt 库来校验用户,于是我重新了一个方法

二、使用步骤

1.新建 channelsmiddleware.py 文件

修改asgi:

# mysite/asgi.pyimport osfrom mysite.channelsmiddlewareimport JwtAuthMiddlewarefrom channels.routingimport ProtocolTypeRouter, URLRouterfrom django.core.asgiimport get_asgi_applicationimport chat.routing

os.environ.setdefault("DJANGO_SETTINGS_MODULE","mysite.settings")

application= ProtocolTypeRouter({"http": get_asgi_application(),"websocket": JwtAuthMiddleware(
        URLRouter(
            chat.routing.websocket_urlpatterns)),})

1.往 channelsmiddleware.py 添加

# mysite/channelsmiddleware.pyfrom channels.authimport AuthMiddlewareStackfrom channels.dbimport database_sync_to_asyncfrom channels.middlewareimport BaseMiddlewarefrom django.confimport settingsfrom django.contrib.authimport get_user_modelfrom django.contrib.auth.modelsimport AnonymousUser, Userfrom django.dbimport close_old_connectionsfrom jwtimport decodeas jwt_decodefrom rest_framework_simplejwt.authenticationimport AUTH_HEADER_TYPE_BYTESfrom rest_framework_simplejwt.exceptionsimport InvalidToken, TokenErrorfrom rest_framework_simplejwt.tokensimport UntypedToken@database_sync_to_asyncdefget_user(validated_token):try:
        user= get_user_model().objects.get(id=validated_token["user_id"])# return get_user_model().objects.get(id=toke_id)return userexcept User.DoesNotExist:return AnonymousUser()classJwtAuthMiddleware(BaseMiddleware):def__init__(self, inner):
        self.inner= innerasyncdef__call__(self, scope, receive, send):# Close old database connections to prevent usage of timed out connections
        close_old_connections()
        parts=dict(scope['headers']).get(b'authorization',b'').split()iflen(parts)==0:# Empty AUTHORIZATION header sentreturnNoneif parts[0]notin AUTH_HEADER_TYPE_BYTES:# Assume the header does not contain a JSON web tokenreturnNoneiflen(parts)!=2:raiseNone

        token= parts[1]# Get the token# Try to authenticate the usertry:# This will automatically validate the token and raise an error if token is invalid
            UntypedToken(token)except(InvalidToken, TokenError)as e:# Token is invalidprint(e)returnNoneelse:#  Then token is valid, decode it
            decoded_data= jwt_decode(token, settings.SECRET_KEY, algorithms=["HS256"])print(decoded_data)# Will return a dictionary like -# {#     "token_type": "access",#     "exp": 1568770772,#     "jti": "5c15e80d65b04c20ad34d77b6703251b",#     "user_id": 6# }# Get the user using ID
            scope["user"]=await get_user(validated_token=decoded_data)returnawaitsuper().__call__(scope, receive, send)defJwtAuthMiddlewareStack(inner):return JwtAuthMiddleware(AuthMiddlewareStack(inner))

现在就可以在channels里通过 self.scope[“user”] 获取到一个用户实例了。

总结

以上就是我的方法,多看看源代码,找找资料,其实也不是很难。

  • 作者:Python_anning
  • 原文链接:https://blog.csdn.net/Python_anning/article/details/123868307
    更新时间:2022-08-13 14:05:53