Django Channels框架是否支持编写自定义权限连接特定Consumers？

Django Channels 确实支持自定义权限控制来限制客户端连接特定 Consumer，实现思路和 DRF 类似，但因为是处理长连接（比如 WebSocket），所以需要在连接建立阶段做权限校验。

步骤1：先搞定用户认证 #

和 DRF 的 authentication_classes 作用一样，Channels 自带的 AuthMiddlewareStack 可以把 Django 的用户认证信息挂载到连接的 scope 对象里。你只需要在路由配置里用它包装 WebSocket 路由即可：

from channels.routing import ProtocolTypeRouter, URLRouter
from channels.auth import AuthMiddlewareStack
import myapp.routing

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

这样后续在 Consumer 的 scope 里就能拿到 user 对象，和普通 Django 请求里的用户一致。

步骤2：编写自定义权限类 #

和 DRF 一样，你可以基于 BasePermission 写自己的权限类，核心是实现 has_permission 方法，接收 scope 参数并返回布尔值表示权限是否通过：

from channels.permissions import BasePermission

# 基础的登录权限类
class IsAuthenticated(BasePermission):
    def has_permission(self, scope):
        return scope["user"].is_authenticated

# 自定义权限类示例：检查用户是否在指定组
class MyCustomPermission(BasePermission):
    def has_permission(self, scope):
        return scope["user"].groups.filter(name="websocket_allowed").exists()

步骤3：在 Consumer 中使用权限类 #

在 Consumer 里定义 permission_classes 属性，然后在 connect 方法里逐一校验权限——只要有一个权限不通过，就直接关闭连接：

异步 Consumer 示例 #

from channels.generic.websocket import AsyncWebsocketConsumer

class MyAsyncConsumer(AsyncWebsocketConsumer):
    permission_classes = [IsAuthenticated, MyCustomPermission]

    async def connect(self):
        # 遍历权限类，逐个检查
        for permission_cls in self.permission_classes:
            if not permission_cls().has_permission(self.scope):
                await self.close()
                return
        # 所有权限通过，接受连接
        await self.accept()

    async def disconnect(self, close_code):
        # 断开连接的清理逻辑
        pass

同步 Consumer 示例 #

from channels.generic.websocket import WebsocketConsumer

class MySyncConsumer(WebsocketConsumer):
    permission_classes = [IsAuthenticated, MyCustomPermission]

    def connect(self):
        for permission_cls in self.permission_classes:
            if not permission_cls().has_permission(self.scope):
                self.close()
                return
        self.accept()

这种实现方式完全贴合 DRF 的使用习惯，是 Channels 官方推荐的权限控制方案，能灵活实现各种连接级别的权限校验。

内容的提问来源于stack exchange，提问作者Kassab