DocsPricingBlog
Contact Us
Log inSign up for FREE
Back to Tech Articles

Django ChannelsとFastAPIでのユーザー認証によるWebSocket接続の保護

Sep 21, 2025
# Misc

Ethan Miller

Product Engineer · Leapcell

This post is also available in English, 한국어, and Deutsch
Cover of "Django ChannelsとFastAPIでのユーザー認証によるWebSocket接続の保護"

はじめに：リアルタイムインタラクションの保護

今日の相互接続された世界では、チャットプラットフォーム、共同作業ツール、ライブダッシュボード、ストリーミングサービスなど、あらゆるものを駆動するリアルタイムアプリケーションがユビキタスです。WebSocketはこれらのアプリケーションの基盤であり、クライアントとサーバー間の永続的で双方向の通信を可能にします。しかし、この強力な機能は、以下のような重要なセキュリティ上の懸念をもたらします。厳選されたユーザーだけがリアルタイム機能にアクセスして操作できることをどのように保証しますか？認証されていないWebSocket接続は、データ漏洩、不正アクセス、ユーザーエクスペリエンスの低下につながる可能性があります。この記事では、特に人気のあるPythonフレームワークであるDjango ChannelsとFastAPI内で、WebSocket接続に堅牢なユーザー認証を追加するという不可欠なプロセスを掘り下げ、リアルタイムインタラクションを保護するために必要なツールを提供します。

安全なリアルタイム通信の柱を理解する

実装の詳細に入る前に、議論の基盤となるいくつかのコアコンセプトを明確にしましょう：

  • WebSocket: 単一のTCP接続を介してフルデュプレックス通信チャネルを提供する通信プロトコル。従来のHTTPとは異なり、WebSocketはオープン接続を維持し、繰り返しハンドシェイクなしで、即時で双方向のデータ交換を可能にします。
  • 認証: ユーザーまたはクライアントのIDを確認するプロセス。WebSocketのコンテキストでは、これは接続されているクライアントが主張している本人であることを確認することを意味します。
  • 認可: 認証されたユーザーが何を行うことができるかを決定するプロセス。認証の成功後、認可は特定のリアルタイムリソースまたは機能へのアクセスレベルを決定します。
  • Django Channels: WebSockets、チャットプロトコル、IoTプロトコルなどを処理するためにDjangoの機能を拡張する公式Djangoプロジェクト。DjangoのORMおよび認証システムとシームレスに統合されます。
  • FastAPI: Python 3.7+に基づき、標準Python型ヒントを利用したAPI構築のためのモダンで高速（高性能）なWebフレームワーク。その速度と非同期機能で知られており、WebSocketアプリケーションに適しています。
  • ASGI（Asynchronous Server Gateway Interface）: WSGIの精神的な後継者であるASGIは、非同期対応のPython Webサーバー、フレームワーク、およびアプリケーション間の標準インターフェイスを提供します。Django ChannelsとFastAPIの両方がASGIを利用しています。

WebSocketのためのユーザー認証の実装

WebSocket接続に認証を追加することは、主に接続ハンドシェイクをインターセプトし、接続が確立される前またはメッセージが交換される前にユーザーの資格情報を検証することを含みます。Django ChannelsとFastAPIのアーキテクチャの違いにより、方法はわずかに異なります。

Django Channelsでの認証

Django ChannelsはDjangoの既存の認証システムと密接に統合されています。典型的なアプローチは、Djangoのセッションベース認証またはトークンベース認証を利用することです。

セッション認証の使用

Djangoアプリケーションが既にHTTPリクエストにセッションベース認証を使用している場合、WebSocketへの拡張は簡単です。Channelsが提供するAuthMiddlewareStackは、WebSocketハンドシェイクに有効なセッションIDが存在する場合、認証されたユーザーでscope['user']を自動的に入力できます。

まず、WebSocket用のルーティング（asgi.pyまたはルーティング構成）にAuthMiddlewareStackが含まれていることを確認してください。

# your_project/asgi.py

import os

from channels.auth import AuthMiddlewareStack
from channels.routing import ProtocolTypeRouter, URLRouter
from django.core.asgi import get_asgi_application

from my_app import routing  # （WebSocket用のrouting.pyがあると仮定）

os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'your_project.settings')

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

次に、コンシューマー内で、self.scope['user']オブジェクトに認証されたDjangoユーザーインスタンスが含まれます。ユーザーが認証されていない場合、self.scope['user']はAnonymousUserインスタンスになります。

# my_app/consumers.py

import json
from channels.generic.websocket import AsyncWebsocketConsumer

class MyChatConsumer(AsyncWebsocketConsumer):
    async def connect(self):
        # ユーザーが認証されているか確認
        if self.scope['user'].is_authenticated:
            self.room_name = self.scope['url_route']['kwargs']['room_name']
            self.room_group_name = 'chat_%s' % self.room_name

            # ルームグループに参加
            await self.channel_layer.group_add(
                self.room_group_name,
                self.channel_name
            )
            await self.accept()
            await self.send(text_data=json.dumps({
                'message': f"Welcome, {self.scope['user'].username}!"
            }))
        else:
            # 認証されていない場合は接続を拒否
            await self.close(code=4003) # 認証されていない場合のカスタムクローズコード
            print("WebSocket connection rejected: User not authenticated.")

    async def disconnect(self, close_code):
        if self.scope['user'].is_authenticated:
            # ルームグループから退出
            await self.channel_layer.group_discard(
                self.room_group_name,
                self.channel_name
            )

    async def receive(self, text_data):
        if self.scope['user'].is_authenticated:
            text_data_json = json.loads(text_data)
            message = text_data_json['message']

            # ルームグループにメッセージを送信
            await self.channel_layer.group_send(
                self.room_group_name,
                {
                    'type': 'chat_message',
                    'message': message,
                    'username': self.scope['user'].username
                }
            )
        else:
            await self.send(text_data=json.dumps({
                'error': 'You must be logged in to send messages.'
            }))

    async def chat_message(self, event):
        message = event['message']
        username = event['username']

        # WebSocketにメッセージを送信
        await self.send(text_data=json.dumps({
            'message': message,
            'username': username
        }))

クライアントからの接続時、ブラウザは自動的にセッションCookieを送信し、Django ChannelsのAuthMiddlewareStackは認証に使用します。

トークン認証の使用

API駆動型アプリケーションやブラウザセッションがないシナリオでは、トークンベース認証（例：JWT）が好まれることがよくあります。カスタム認証ミドルウェアを作成する必要があります。クライアントは通常、WebSocket接続URLでクエリパラメータとして、またはカスタムヘッダー（ただし、WebSocketハンドシェイクでヘッダーを扱うのは難しい場合があります）でトークンを送信します。

カスタムミドルウェアは以下のようになります：

# my_app/token_auth_middleware.py

from channels.db import database_sync_to_async
from django.contrib.auth.models import AnonymousUser
from rest_framework_simplejwt.authentication import JWTAuthentication
from rest_framework_simplejwt.exceptions import InvalidToken, TokenError

@database_sync_to_async
def get_user_from_token(token):
    # この関数はJWTライブラリに基づいて適応させる必要があります
    # 例：djangorestframework-simplejwtを使用
    try:
        validated_token = JWTAuthentication().get_validated_token(token)
        user = JWTAuthentication().get_user(validated_token)
        return user
    except (InvalidToken, TokenError):
        return AnonymousUser()

class TokenAuthMiddleware:
    def __init__(self, app):
        self.app = app

    async def __call__(self, scope, receive, send):
        try:
            # クエリパラメータからトークンを抽出（例：ws://localhost/ws/chat/?token=YOUR_TOKEN）
            query_string = scope['query_string'].decode()
            query_params = dict(qp.split("=") for qp in query_string.split("&") if "=" in qp)
            token = query_params.get("token")

            if token:
                scope['user'] = await get_user_from_token(token)
            else:
                scope['user'] = AnonymousUser()
        except ValueError: # クエリ文字列が完全に形式化されていない場合を処理
            scope['user'] = AnonymousUser()

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

# your asgi.pyで：
# application = ProtocolTypeRouter({
#     "http": get_asgi_application(),
#     "websocket": TokenAuthMiddleware(  # カスタムミドルウェアを使用
#         URLRouter(
#             routing.websocket_urlpatterns
#         )
#     ),
# })

クライアントは ws://localhost:8000/ws/chat/room_slug/?token=YOUR_JWT_TOKEN のように接続します。

FastAPIでの認証

ASGIフレームワークであるFastAPIは、WebSocketの認証を処理するための柔軟な方法を提供しており、しばしばその強力な依存性注入システムを活用します。最も一般的なアプローチは、接続ハンドシェイク中にWebSocket接続URLまたはカスタムヘッダーからトークンを抽出することです。

# main.py

from fastapi import FastAPI, WebSocket, WebSocketDisconnect, Depends, status
from fastapi.security import OAuth2PasswordBearer
from jose import JWTError, jwt
from pydantic import BaseModel
from typing import Dict, Any

# モックユーザーデータベースとJWT設定（実際のインプリメンテーションに置き換えてください）
SECRET_KEY = "your-secret-key" # 本番環境では、環境変数を使用
ALGORITHM = "HS256"

class UserInDB(BaseModel):
    username: str
    email: str | None = None
    full_name: str | None = None
    disabled: bool | None = None

# ユーザー取得のための非常に基本的なモック
async def get_user_from_db(username: str):
    if username == "testuser":
        return UserInDB(username="testuser", email="test@example.com")
    return None

app = FastAPI()

# OAuth2PasswordBearerは主にHTTP用ですが、その一部のロジックを適応させることができます。
# WebSocketでは、トークンを直接抽出します。
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token") # プレースホルダー、WSパスには直接使用されません

async def authenticate_websocket_user(websocket: WebSocket, token: str | None = None):
    if not token:
        await websocket.close(code=status.WS_1008_POLICY_VIOLATION) # "Policy Violation"でクローズ
        raise WebSocketDisconnect("Authentication token missing")

    try:
        payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
        username: str = payload.get("sub")
        if username is None:
            await websocket.close(code=status.WS_1008_POLICY_VIOLATION)
            raise WebSocketDisconnect("Could not validate credentials")
        user = await get_user_from_db(username) # 実際のユーザー取得ロジックに置き換えてください
        if user is None:
            await websocket.close(code=status.WS_1008_POLICY_VIOLATION)
            raise WebSocketDisconnect("User not found")
        return user
    except JWTError:
        await websocket.close(code=status.WS_1008_POLICY_VIOLATION)
        raise WebSocketDisconnect("Invalid authentication token")

@app.websocket("/ws/chat/{room_id}")
async def websocket_endpoint(
    websocket: WebSocket,
    room_id: str,
    # クエリパラメータまたはカスタムヘッダーからトークンを抽出
    # クエリパラメータの場合：
    token: str = None, # 'token'をクエリパラメータにする
    # ヘッダーの場合（カスタムクライアントコードが必要、例：JavaScriptで）：
    # sec_websocket_protocol: str | None = Header(None, alias="sec-websocket-protocol"),

):
    # クエリパラメータを使用する場合、$token$はパスパラメータとして直接利用可能になります。
    # sec-websocket-protocol ヘッダーを使用する場合、$token$を抽出するために解析します。例：「token, YOUR_JWT_TOKEN」
    # 簡単にするため、トークンはクエリパラメータまたは依存関係として直接渡されると仮定します。
    
    try:
        # ユーザーを認証する
        current_user = await authenticate_websocket_user(websocket, token=token)
        print(f"User {current_user.username} authenticated for room {room_id}")

        await websocket.accept()
        await websocket.send_json({"message": f"Welcome, {current_user.username}! You are in room {room_id}."})
        while True:
            data = await websocket.receive_text()
            await websocket.send_text(f"Message from {current_user.username}: {data}")
    except WebSocketDisconnect as e:
        print(f"WebSocket disconnected for user {current_user.username if 'current_user' in locals() else 'unauthenticated'}: {e}")
    except Exception as e:
        print(f"An error occurred: {e}")

このFastAPIの例では：

  1. authenticate_websocket_userをWebSocketオブジェクトとtoken文字列を受け取るasync関数として定義します。
  2. この関数内で、JWTトークンのデコードを試みます。トークンが無効であるか、ユーザーが見つからない場合、特定のステータスコード（WS_1008_POLICY_VIOLATION）でWebSocket接続をcloseし、WebSocketDisconnect例外を発生させます。
  3. メインのwebsocket_endpoint関数は、クエリパラメータ（例：ws://localhost:8000/ws/chat/123?token=YOUR_JWT）としてtokenを受け取ることができます。
  4. current_userは、authenticate_websocket_userを介して暗黙的に渡され、返されるため、WebSocket内の後続の操作でユーザーを識別できます。

クライアントは ws://localhost:8000/ws/chat/myroom?token=YOUR_JWT_TOKEN のように接続します。

トークン送信のためのクライアントサイドの考慮事項

トークン認証（特にJWT）を使用する場合、クライアントは明示的にトークンを送信する必要があります。

Django Channels（トークン認証）およびFastAPIの場合：

  • クエリパラメータ: 最も簡単な方法です。クライアントはWebSocket URLにトークンを付加します： 
    const token = localStorage.getItem('access_token'); // またはCookieから取得
const ws = new WebSocket(`ws://localhost:8000/ws/chat/${roomId}/?token=${token}`);
  • カスタムヘッダー: より安全なトークン（機密トークン）の場合、一部のWebSocket実装ではハンドシェイク中にカスタムヘッダーを簡単に許可できない場合があります。クライアントがそれをサポートしている場合（例：Node.jsのwsライブラリ、一部のブラウザ拡張機能がこれを許可します）、カスタムヘッダーを定義できます。ブラウザでは、通常、WebSocketにアップグレードする前にsockjsのようなライブラリを使用するか、XHRベースのハンドシェイクを手動で管理する必要があります。ブラウザベースのJWT（WebSocket用）にはクエリパラメータを使用するのが最も簡単です。 
    // これはブラウザベースのJavaScript WebSocketではより複雑です
// 通常、SockJSラッパーを使用するか、XHRベースのハンドシェイクを手動で管理する場合にうまく機能します。
const ws = new WebSocket(`ws://localhost:8000/ws/chat/${roomId}`, ['protocol', 'token,YOUR_JWT_TOKEN']);

アプリケーションシナリオ

  • リアルタイムチャットアプリケーション: 認証されたユーザーのみが特定のチャットルームでメッセージを送受信できます。
  • ライブダッシュボード: 認証されたユーザーの権限に合わせて調整された機密性の高い分析データを表示します。
  • 共同編集: 許可されたチームメンバーのみがリアルタイムでドキュメントを編集できることを保証します。
  • 通知システム: 個々の認証されたユーザーにパーソナライズされた通知を送信します。
  • ゲーミング: ゲームセッションに参加を許可する前にプレイヤーのIDを検証します。

結論：リアルタイムフロンティアの強化

ユーザー認証によるWebSocket接続の保護は、単なるベストプラクティスではなく、堅牢で信頼性が高く、信頼できるリアルタイムアプリケーションを構築するための基本的な要件です。Django ChannelsのDjango認証システムとのシームレスな統合を利用する場合でも、JWTベースのアプローチのためのFastAPIの柔軟な依存性注入を利用する場合でも、原則は一貫しています。接続ハンドシェイクでIDを確認することです。適切な認証を実装することで、ユーザーデータを保護し、リアルタイム機能へのアクセスを制御し、すべての人にとってより安全なデジタルエクスペリエンスを構築できます。WebSocketの認証は、安全で管理されたリアルタイム環境を構築するための鍵です。

Share this article

HackerNews

More Posts from Leapcell

Popular Posts