2025-01-04
Django Channels , , , , , ,

Sessions、身份驗證、安全性 – Django Channels：實現 WebSocket 異步通信，打造高效的長連接系統

A. Sessions

Channels 支援使用 HTTP Cookie 的標準 Django 會話，這適用於 HTTP 和 WebSocket。不過，有一些注意事項需要留意。

1. 基本用法

Django Channels 的 SessionMiddleware 支援標準的 Django sessions，像所有中介軟體一樣，應該包裹在需要 session 資訊的 ASGI 應用程式上（例如，將其應用到整個消費者集合的 URLRouter，或是單個消費者）。SessionMiddleware 需要搭配 CookieMiddleware 才能運作。為了方便起見，這些也已經合併為一個名為 SessionMiddlewareStack 的 callable，包含了兩者。這些都可以從 channels.session 匯入。

要使用這個中介軟體，需要再在 asgi.py 中將其包裹在適當層級的消費者周圍。

from channels.routing import ProtocolTypeRouter, URLRouter
from channels.security.websocket import AllowedHostsOriginValidator
from channels.sessions import SessionMiddlewareStack

from myapp import consumers

application = ProtocolTypeRouter({
    "websocket": AllowedHostsOriginValidator(
        SessionMiddlewareStack(
            URLRouter([
                path("frontend/", consumers.AsyncChatConsumer.as_asgi()),
            ])
        )
    ),

})

SessionMiddleware 只會在範圍內提供 HTTP 標頭的協議上運作——預設情況下，這包括 HTTP 和 WebSocket。

要在你的消費者（consumer）代碼中使用會話，可以使用 self.scope[“session”]。

class ChatConsumer(WebsocketConsumer):
    def connect(self, event):
        self.scope["session"]["seed"] = random.randint(1, 1000)

SessionMiddleware 遵循與預設的 Django 會話框架相同的所有 Django 設定，例如 SESSION_COOKIE_NAME 及 SESSION_COOKIE_DOMAIN。

2. Session保存

在 HTTP consumers 或 ASGI 應用中，會話的持久性如同你在 Django HTTP 視圖中期望的一樣運作——只要你傳送不具備狀態碼500的 HTTP 回應時，會話就會被保存。這是通過覆蓋任何 http.response.start 消息來完成的，以在你發送回應時將 cookie 標頭注入回應中。如果你將 `SESSION_SAVE_EVERY_REQUEST` 設定為 True，那麼它將在每次回應時保存會話並發送 cookie，否則只會在會話被修改時才保存。

然而，如果你處於 WebSocket 消費者中，會話會被填充但不會自動被保存——你必須自己調用 `scope["session"].save()`（或異步版本 `scope["session"].asave()`）來將會話持久化到你的會話存儲中。如果你不保存，會話仍會在消費者內正常工作（因為它被存儲為一個實例變數），但其他連接或 HTTP 視圖不會看到這些更改。

如果您在使用長輪詢的 HTTP 消費者（consumer），您可能需要在發送回應之前保存對會話的更改。如果您想這樣做，請調用 `scope["session"].save()`。

B. 身份驗證

Channels 原生支援標準的 Django 身份驗證，適用於 HTTP 和 WebSocket 消費者。如果您想支援不同的身份驗證方案（例如，URL 中的令牌），您可以撰寫自己的中間件或處理代碼。

1. Django authentication

Channels中的AuthMiddleware支持標準的Django身份驗證，其中使用者詳情存儲在會話中。它允許在範圍內以唯讀方式訪問user對象。 

AuthMiddleware需要SessionMiddleware才能運作，而SessionMiddleware本身需要CookieMiddleware。為了方便起見，這些都被作為一個名為AuthMiddlewareStack的組合呼叫提供，其中包含了這三者。 

要使用中介軟體，請將它包裹在asgi.py中相應層級的消費者周圍：

from django.urls import re_path

from channels.routing import ProtocolTypeRouter, URLRouter
from channels.auth import AuthMiddlewareStack
from channels.security.websocket import AllowedHostsOriginValidator

from myapp import consumers

application = ProtocolTypeRouter({
    "websocket": AllowedHostsOriginValidator(
        AuthMiddlewareStack(
            URLRouter([
                re_path(r"^front(end)/$", consumers.AsyncChatConsumer.as_asgi()),
            ])
        )
    ),

})

雖然您可以將中介軟體（middleware）包裝在每個消費者（consumer）上，但建議您將其包裝在較高層級的應用程式組件上，例如在這個例子中使用 URLRouter。 

請注意，AuthMiddleware 僅適用於在其作用域中提供 HTTP 標頭的協議——默認情況下，這適用於 HTTP 和 WebSocket。 

要訪問用戶，只需在您的消費者代碼中使用 `self.scope["user"]` 即可。

class ChatConsumer(WebsocketConsumer):
    def connect(self):
        self.user = self.scope["user"]
        self.accept()

2. Custom Authentication

如果您有自訂的身份驗證方案，您可以撰寫一個自訂中介軟體來解析細節，並將使用者物件（或任何其他您需要的物件）放入您的 scope 中。中介軟體被寫成一個可呼叫物件，它接受一個 ASGI 應用程式並將其包裝以回傳另一個 ASGI 應用程式。大多數的身份驗證可以僅在 scope 上完成，所以您所需做的只是覆蓋接收 scope 的初始建構函式，而非事件運行的協程。下面是一個中介軟體的簡單示例，只是從查詢字符串中取出使用者 ID 並加以利用：

from channels.db import database_sync_to_async

@database_sync_to_async
def get_user(user_id):
    try:
        return User.objects.get(id=user_id)
    except User.DoesNotExist:
        return AnonymousUser()

class QueryAuthMiddleware:
    """
    Custom middleware (insecure) that takes user IDs from the query string.
    """

    def __init__(self, app):
        # Store the ASGI application we were passed
        self.app = app

    async def __call__(self, scope, receive, send):
        # Look up user from query string (you should also do things like
        # checking if it is a valid user ID, or if scope["user"] is already
        # populated).
        scope['user'] = await get_user(int(scope["query_string"]))

        return await self.app(scope, receive, send)

相同的原則可以應用於非 HTTP 協議的身份驗證；例如，您可能希望使用聊天協議中的某人的聊天用戶名來將其轉換為用戶。

3. How to log a user in/out

Channels 提供了直接的登入和登出功能（就像 Django 的 contrib.auth 套件一樣）作為 channels.auth.login 和 channels.auth.logout。在您的消費者（consumer）中，您可以使用 await login(scope, user, backend=None) 來登入一個使用者。這要求您的 scope 中有一個 session 物件；最好的方法是確保您的消費者被包裹在 SessionMiddlewareStack 或 AuthMiddlewareStack 中。您可以使用 async 函數 logout(scope) 來登出一個使用者。

如果您是在 WebSocket 消費者中，或者是在 http 消費者中發送第一次響應後登入，則 session 會被填充，但不會自動儲存——您必須在您的消費者代碼中登入後調用 scope["session"].save()。

from channels.auth import login

class ChatConsumer(AsyncWebsocketConsumer):

    ...

    async def receive(self, text_data):
        ...
        # login the user to this session.
        await login(self.scope, user)
        # save the session (if the session backend does not access the db you can use `sync_to_async`)
        await database_sync_to_async(self.scope["session"].save)()

當從同步函式中調用 `login(scope, user)`、`logout(scope)` 或 `get_user(scope)` 時，您需要將它們包裝在 `async_to_sync` 中，因為我們僅提供異步版本：

from asgiref.sync import async_to_sync
from channels.auth import login

class SyncChatConsumer(WebsocketConsumer):

    ...

    def receive(self, text_data):
        ...
        async_to_sync(login)(self.scope, user)
        self.scope["session"].save()

如果您正在使用長時間運行的消費者、WebSocket 或長輪詢 HTTP，可能會發生在您的消費者運行期間用戶在其他地方登出了他們的會話。您可以定期使用 `get_user(scope)` 來確保用戶仍然處於登入狀態。

4. 社群專案 – Django Channels auth token middlewares

在官方文件上列出的Community Projects，基於Django Channels進行開發，其中channels-auth-token-middlewares 是 Django Channels 的一個中介軟體，用於處理身份驗證。這包括使用 Django REST framework 的 token authentication middleware 和 SimpleJWT 的中介軟體，如 QueryStringSimpleJWTAuthTokenMiddleware，為 WebSocket 的身份驗證而設計。

使用時先執行指令安裝Python套件，pip install channels-auth-token-middlewares，並在settings的INSTALLED_APPS中加入channels_auth_token_middlewares。

以下節錄幾個用法範例：

from channels.routing import ProtocolTypeRouter, URLRouter

from channels_auth_token_middlewares.middleware import DRFAuthTokenMiddleware


application = ProtocolTypeRouter({
    "websocket": DRFAuthTokenMiddleware(
        URLRouter([
            # app paths
        ]),
    ),

})

from channels.routing import ProtocolTypeRouter, URLRouter

from channels_auth_token_middlewares.middleware import SimpleJWTAuthTokenMiddleware


application = ProtocolTypeRouter({
    "websocket": SimpleJWTAuthTokenMiddleware(
        URLRouter([
            # app paths
        ]),
    ),

})

詳細支援的驗證方式可參考channels-auth-token-middlewares專案的Github，專案最後更新為兩年前，如果使用的驗證方式較為特定，可以考慮自行開發middleware。

C. 安全性

涵蓋了通過 Channels 提供的協議的基本安全性以及我們提供的輔助工具。

WebSockets 在初始階段是一個 HTTP 請求，包括所有的 cookies 和標頭，因此您可以使用標準的身份驗證代碼來獲取當前會話並檢查用戶 ID。不過，WebSockets 存在跨站請求偽造（CSRF）的風險，因為它們可以從互聯網上的任何站點啟動到您的域，並且仍然會持有來自您網站的用戶 cookies 和會話。如果您通過套接字傳送私人數據，您應該限制允許打開套接字的網站。

這可以通過 `channels.security.websocket` 套件及其包含的兩個 ASGI 中介軟體進行：`OriginValidator` 和 `AllowedHostsOriginValidator`。`OriginValidator` 允許您限制每個 WebSocket 發送的 Origin 標頭的合法選項，以指出其來源。只需將它包裹在您的 WebSocket 應用程式代碼周圍，並將有效域名的列表作為第二個參數傳遞給它。您可以僅傳遞一個域名（例如，.allowed-domain.com）或一個完整的來源，以 scheme://domain[:port] 的格式（例如，http://allowed-domain.com:80）進行傳遞。Port 是可選的，但建議使用。

from channels.security.websocket import OriginValidator

application = ProtocolTypeRouter({

    "websocket": OriginValidator(
        AuthMiddlewareStack(
            URLRouter([
                ...
            ])
        ),
        [".goodsite.com", "http://.goodsite.com:80", "http://other.site.com"],
    ),
})

注意：如果您想解析任何域，則使用來源 `*`。

通常，您想要限制的域集與Django的`ALLOWED_HOSTS`設置相同，該設置對Host標頭執行類似的安全檢查。因此，`AllowedHostsOriginValidator`使您可以使用此設置而無需重新宣告。

from channels.security.websocket import AllowedHostsOriginValidator

application = ProtocolTypeRouter({

    "websocket": AllowedHostsOriginValidator(
        AuthMiddlewareStack(
            URLRouter([
                ...
            ])
        ),
    ),
})

AllowedHostsOriginValidator 會自動允許本地連線通過，如果網站處於 DEBUG 模式，這與 Django 的主機驗證類似。

關於CSRF風險，可以參考以下方式進一步保障安全性：

  1. 在WebSocket連接設置中自行檢查CSRF令牌：當WebSocket初始連接進行時，可以將CSRF令牌作為請求的一部分發送，並在服務器端檢查其有效性。
  2. 應用用戶身份驗證：確保WebSocket的使用者已經在HTTP層進行過身份驗證。這可以通過在WebSocket初始化時附加識別資訊（如session密鑰或JWT token），服務器端用以檢查用戶身份的有效性來實現。
  3. 限制WebSocket的作用域與權限：基於用戶的角色或權限限制他們可通過WebSocket訪問的資源和數據。這可以阻止已認證用戶執行未授權的操作。
  4. 嚴格的連接速率限制：通過限制單個用戶短時間內可以開啟的WebSocket連接數量，以及數據發送頻率，從而減少由惡意用戶造成的拒絕服務（DoS）風險。

以上建議僅供參考，請向資安專家諮詢專業建議。

D. 總結

Django Channels 提供了支援異步通訊的功能，同時也在會話、身份驗證與安全性上提供了有效的解決方案。在會話處理方面，Channels 支援標準的 Django 會話，並提供 SessionMiddlewareStack 來方便操作，這使得在 WebSocket 或 HTTP 消費者中使用會話變得更為直觀。身份驗證則透過 AuthMiddleware 來實現，允許在 WebSocket 中輕鬆使用 Django 的用戶系統；同時，開發者也可以自訂中介軟體以支援各種身份驗證方案。安全性方面，Channels 利用 OriginValidator 和 AllowedHostsOriginValidator 來限制 WebSocket 連接的來源，防止跨站請求偽造（CSRF）攻擊。此外，推薦使用 CSRF 檢查、嚴格的用戶驗證及權限控制等措施以進一步保障應用的安全性。這些功能共同確保在強化 Django Applications 時，能夠提供穩健且相對安全的異步通訊支持。

E. 系列文章