用于通道 websocket 身份验证的 Django jwt 中间件

Posted 2023-03-10

【中文标题】用于通道 websocket 身份验证的 Django jwt 中间件

【英文标题】：Django jwt middleware for channels websocket authentication

【发布时间】：2018-11-26 19:46:44

【问题描述】：

我正在尝试为 django 频道设置身份验证中间件。我希望这个中间件只对 websocket 请求有效。

似乎在这种情况下我没有获得完整的中间件功能。例如我无法让response = self.get_response(scope) 工作：

'TokenAuthMiddleware' object has no attribute 'get_response'

现在这个中间件一切正常（它只为 websocket 请求激活，而不是在settings.py 中注册），除了我需要一种方法来修改响应状态代码（阻止匿名用户并为 @ 设置错误代码987654324@)。任何帮助表示赞赏。我使用 Django 2.0.6 和频道 2.1.1。通过djangorestframework-jwt进行jwt认证

中间件：

import jwt, re
import traceback
import logging

from channels.auth import AuthMiddlewareStack
from django.contrib.auth.models import AnonymousUser
from django.conf import LazySettings
from jwt import InvalidSignatureError, ExpiredSignatureError, DecodeError

from project.models import MyUser

settings = LazySettings()
logger = logging.getLogger(__name__)


class TokenAuthMiddleware:
    """
    Token authorization middleware for Django Channels 2

    """

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

    def __call__(self, scope):
        headers = dict(scope['headers'])
        auth_header = None
        if b'authorization' in headers:
            auth_header = headers[b'authorization'].decode()
        else:
            try:
                auth_header = _str_to_dict(headers[b'cookie'].decode())['X-Authorization']
            except:
                pass

        logger.info(auth_header)

        if auth_header:
            try:
                user_jwt = jwt.decode(
                    auth_header,
                    settings.SECRET_KEY,
                )
                scope['user'] = MyUser.objects.get(
                    id=user_jwt['user_id']
                )
            except (InvalidSignatureError, KeyError, ExpiredSignatureError, DecodeError):
                traceback.print_exc()
                pass
            except Exception as e:  # NoQA
                logger.error(scope)
                traceback.print_exc()

        return self.inner(scope)


TokenAuthMiddlewareStack = lambda inner: TokenAuthMiddleware(AuthMiddlewareStack(inner))


def _str_to_dict(str):
    return k: v.strip('"') for k, v in re.findall(r'(\S+)=(".*?"|\S+)', str)

routing.py

application = ProtocolTypeRouter(
    # (http->django views is added by default)
    'websocket': TokenAuthMiddlewareStack(
        URLRouter(
            cmonitorserv.routing.websocket_urlpatterns
        )
    ),
)

【参考方案1】：

无法使用中间件找到解决方案。 现在通过处理consumers.py中的身份验证权限解决了

def _is_authenticated(self):
    if hasattr(self.scope, 'auth_error'):
        return False
    if not self.scope['user'] or self.scope['user'] is AnonymousUser:
        return False
    return True

另一个似乎没有在任何地方记录的重要事情 - 要拒绝与自定义错误代码的连接，我们需要先接受它。

class WebConsumer(WebsocketConsumer):
    def connect(self):
        self.accept()
        if self._is_authenticated():
              ....
        else:
            logger.error("ws client auth error")
            self.close(code=4003)