ドキュメント価格ブログ
お問い合わせ
ログイン無料で新規登録
Back to Tech Articles

Django Channels와 FastAPI에서 사용자 인증으로 WebSocket 연결 보안

9月 21, 2025
# Misc

Ethan Miller

Product Engineer · Leapcell

この記事はEnglish、日本語、Deutschでも利用できます
Cover of "Django Channels와 FastAPI에서 사용자 인증으로 WebSocket 연결 보안"

소개: 실시간 상호작용 보호

오늘날 상호 연결된 세상에서 실시간 애플리케이션은 채팅 플랫폼, 협업 도구부터 라이브 대시보드 및 스트리밍 서비스에 이르기까지 거의 모든 것을 지원하는 기반 기술입니다. WebSocket은 이러한 애플리케이션의 초석으로, 클라이언트와 서버 간의 지속적이고 양방향 통신을 가능하게 합니다. 그러나 이 강력한 기능은 중요한 보안 문제를 야기합니다. 승인된 사용자만 실시간 기능에 액세스하고 상호 작용할 수 있도록 어떻게 보장할 수 있을까요? 인증되지 않은 WebSocket 연결은 데이터 유출, 무단 액세스 및 사용자 경험 침해로 이어질 수 있습니다. 이 글에서는 인기 있는 Python 프레임워크인 Django Channels 및 FastAPI 내에서 WebSocket 연결에 강력한 사용자 인증을 추가하는 필수 프로세스를 자세히 살펴보고 실시간 상호 작용을 보호하는 데 필요한 도구를 제공합니다.

안전한 실시간 통신의 핵심 이해

구현 세부 정보에 들어가기 전에 논의의 기초가 되는 몇 가지 핵심 개념을 명확히 해보겠습니다.

  • WebSocket: 단일 TCP 연결을 통해 전이중 통신 채널을 제공하는 통신 프로토콜입니다. 기존 HTTP와 달리 WebSocket은 연결을 유지하여 반복적인 핸드셰이크 없이 즉각적이고 양방향 데이터 교환을 허용합니다.
  • 인증(Authentication): 사용자 또는 클라이언트의 신원을 확인하는 프로세스입니다. WebSocket의 맥락에서 이는 연결된 클라이언트가 자신이 주장하는 사람임을 보장하는 것을 의미합니다.
  • 인가(Authorization): 인증된 사용자가 수행할 수 있는 작업을 결정하는 프로세스입니다. 성공적인 인증 후 인가는 특정 실시간 리소스 또는 기능에 대한 액세스 수준을 결정합니다.
  • Django Channels: Django의 기능을 확장하여 WebSocket, 채팅 프로토콜, IoT 프로토콜 등을 처리하는 공식 Django 프로젝트입니다. Django의 ORM 및 인증 시스템과 원활하게 통합됩니다.
  • FastAPI: Python 3.7+ 기반으로 Python 타입 힌트를 활용하는 현대적이고 빠른(고성능) API 구축용 웹 프레임워크입니다. 속도와 비동기 기능으로 유명하며 WebSocket 애플리케이션에 적합합니다.
  • ASGI (Asynchronous Server Gateway Interface): WSGI의 정신적 후계자인 ASGI는 비동기 기능의 Python 웹 서버, 프레임워크 및 애플리케이션 간의 표준 인터페이스를 제공합니다. Django Channels와 FastAPI 모두 ASGI를 활용합니다.

WebSocket에 대한 사용자 인증 구현

WebSocket 연결에 인증을 추가하는 것은 주로 연결 핸드셰이크를 가로채고 연결이 설정되거나 메시지가 교환되기 전에 사용자 자격 증명을 확인하는 것을 포함합니다. 아키텍처 차이로 인해 Django Channels와 FastAPI 간의 방법은 약간 다릅니다.

Django Channels에서의 인증

Django Channels는 Django의 기존 인증 시스템과 긴밀하게 통합됩니다. 일반적인 접근 방식은 Django의 세션 기반 인증 또는 토큰 기반 인증을 활용하는 것입니다.

세션 인증 사용

Django 애플리케이션이 이미 HTTP 요청에 대해 세션 기반 인증을 사용하는 경우 WebSocket으로 확장하는 것은 간단합니다. Channels에서 제공하는 AuthMiddlewareStack은 유효한 세션 ID가 WebSocket 핸드셰이크에 있는 경우 인증된 사용자로 scope['user']를 자동으로 채울 수 있습니다.

먼저 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
        }))

클라이언트에서 연결할 때 브라우저는 자동으로 세션 쿠키를 전송하며, 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)

# 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에서의 인증

FastAPI는 ASGI 프레임워크로서 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) # "정책 위반"으로 닫기
        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"),
):
    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 함수는 token을 쿼리 매개변수(예: ws://localhost:8000/ws/chat/123?token=YOUR_JWT)로 받을 수 있습니다.
  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'); // 또는 쿠키에서 가져오기
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']);

애플리케이션 시나리오

  • 실시간 채팅 애플리케이션: 인증된 사용자만 특정 채팅방에서 메시지를 보내고 받을 수 있습니다.
  • 라이브 대시보드: 인증된 사용자의 권한에 맞춰 조정된 민감한 분석 데이터를 표시합니다.
  • 협업 편집: 인증된 팀원만 문서를 실시간으로 수정할 수 있도록 보장합니다.
  • 알림 시스템: 개별 인증된 사용자에게 개인화된 알림을 보냅니다.
  • 게임: 게임 세션에 참여하기 전에 플레이어의 신원을 확인합니다.

결론: 실시간 프론티어 강화

사용자 인증을 통한 WebSocket 연결 보안은 단순한 모범 사례가 아니라 강력하고 안정적이며 신뢰할 수 있는 실시간 애플리케이션을 구축하기 위한 기본 요구 사항입니다. Django Channels가 Django의 인증 시스템과 원활하게 통합되는지 여부와 관계없이 또는 JWT 기반 접근 방식에 대한 FastAPI의 유연한 의존성 주입을 활용하는지 여부에 관계없이 원칙은 일관됩니다. 연결 핸드셰이크에서 신원을 확인하십시오. 적절한 인증을 구현함으로써 사용자의 데이터를 보호하고, 실시간 기능에 대한 액세스를 제어하며, 모든 사람을 위한 더 안전한 디지털 경험을 구축할 수 있습니다. WebSocket을 인증하는 것은 안전하고 제어된 실시간 환경을 구축하는 데 중요합니다.

Share this article

HackerNews

More Posts from Leapcell

Popular Posts