### 在 Django 中为 WebSocket 添加拦截器实现访问授权 在 Django 中，WebSocket 的支持通常通过与 Channels 库结合使用来实现。Channels 是一个用于扩展 Django 以支持异步功能（如 WebSocket、长轮询等）的库。为了在 WebSocket 连接中实现访问授权，可以通过编写自定义的中间件来拦截和验证连接请求。 以下是实现此功能的具体方法： #### 1. 安装并配置 Django Channels 首先，确保已安装 `channels` 和 `channels-redis`（如果需要 Redis 作为通道层）。然后，在项目的 `settings.py` 文件中进行如下配置[^3]： ```python INSTALLED_APPS = [ # ... 'channels', ] ASGI_APPLICATION = 'your_project_name.asgi.application' CHANNEL_LAYERS = { 'default': { 'BACKEND': 'channels_redis.core.RedisChannelLayer', 'CONFIG': { "hosts": [('127.0.0.1', 6379)], }, }, } ``` #### 2. 创建自定义中间件 为了实现访问授权，可以创建一个自定义的 ASGI 中间件。该中间件将在每个 WebSocket 连接请求上运行，并检查用户是否具有访问权限。 以下是一个示例中间件代码[^4]： ```python # your_project/middleware.py from channels.db import database_sync_to_async from django.contrib.auth.models import AnonymousUser from django.contrib.auth import get_user_model @database_sync_to_async def get_user(scope): User = get_user_model() try: if "token" in scope["headers"]: token_key = scope["headers"]["token"].decode("utf-8") user = User.objects.get(auth_token=token_key) return user except Exception: pass return AnonymousUser() class TokenAuthMiddleware: def __init__(self, inner): self.inner = inner def __call__(self, scope): return TokenAuthMiddlewareInstance(scope, self.inner) class TokenAuthMiddlewareInstance: def __init__(self, scope, inner): self.scope = dict(scope) self.inner = inner async def __call__(self, receive, send): self.scope['user'] = await get_user(self.scope) inner = self.inner(self.scope) return await inner(receive, send) ``` #### 3. 配置 ASGI 应用程序 接下来，将自定义中间件添加到 ASGI 应用程序中。编辑 `asgi.py` 文件以包含中间件[^5]： ```python # your_project/asgi.py import os from django.core.asgi import get_asgi_application from channels.routing import ProtocolTypeRouter, URLRouter from channels.auth import AuthMiddlewareStack from your_project.middleware import TokenAuthMiddleware from your_app import routing os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'your_project.settings') application = ProtocolTypeRouter({ "http": get_asgi_application(), "websocket": TokenAuthMiddleware( AuthMiddlewareStack( URLRouter( routing.websocket_urlpatterns ) ) ), }) ``` #### 4. 定义路由 在应用程序中定义 WebSocket 路由。例如： ```python # your_app/routing.py from django.urls import path from . import consumers websocket_urlpatterns = [ path('ws/some_route/', consumers.MyConsumer.as_asgi()), ] ``` #### 5. 创建消费者 最后，创建 WebSocket 消费者以处理客户端的消息。确保在消费者中检查用户的授权状态[^6]： ```python # your_app/consumers.py from channels.generic.websocket import AsyncWebsocketConsumer class MyConsumer(AsyncWebsocketConsumer): async def connect(self): if self.scope['user'].is_authenticated: await self.accept() else: await self.close() async def disconnect(self, close_code): pass async def receive(self, text_data=None, bytes_data=None): # 处理接收到的消息 pass ``` 通过上述步骤，可以在 Django 中为 WebSocket 实现访问授权的功能。 --- ###