[AINFO] WebSocket JWT 인증 구현

28 Mar 2025 in Dev Log

⚠️ 안내: 이 글은 학습 기록용입니다. 오류나 보완 의견은 댓글로 알려주세요.

📌 관련 개념 정리

관련된 기술에 대한 개념적 정리는 Authentication (Session, Token), Middleware를 참고해주세요.

🧭 배경

AINFO 프로젝트에서는 세션(Session) 기반의 인증 방식을 사용하지 않고 JWT Token 기반의 인증 방식을 사용하고 있었습니다. HTTP API에서는 django-rest-framework-simplejwt를 활용해 비교적 자연스럽게 인증 흐름을 구성할 수 있었고, 이 구조자체에는 문제가 없다고 생각했었습니다.

하지만, WebSocket을 붙인 뒤 확인해보니,

기존 JWT 인증 체계가 쳇봇 화면(WebSocket 환경)에서는 전혀 적용되지 않았고
WebSocket 연결 자체는 누구나 가능한 상태였습니다.

이전 기수들의 포트폴리오를 살펴봐도

Django + JWT + Channels 구조를 사용하고 있음에도
WebSocket 구간에서는 별도의 인증을 하고 있지 않았습니다.

관련 내용이 많이 없고 난이도가 있어보여서 생략할까도 생각해봤지만, WebSocket 구간은 GPT API를 호출하고 실제 비용과 리소스가 발생하는 핵심 기능의 진입점이므로 인증 없이 모든 사람에게 열어둔다는 말이 안된다고 생각했습니다.

보안에 대해 전문적인 지식을 가지고 있는 건 아니지만, “HTTP에서와 최소한 동일한 수준의 인증은 유지하는 게 맞지 않나” 라고 생각을 했고 어떻게 해서든 JWT 인증을 구현하기로 결정했습니다.

🧩 구현 과정

전체적인 코드는 Medium의 글Authentication in WebSocket with Django and Django Rest Framework (DRF) Joseph Miracle을 참고하였습니다.

1. chatbot/middleware.py 추가

1-1. WebSocket 연결 요청에서 JWT 토큰 추출

def get_token_from_scope(self, scope):
    query_string = parse_qs(scope.get("query_string", b"").decode("utf-8"))
    return query_string.get("token", [None])[0]

설명:
- WebSocket 연결은 HTTP handshake로 시작되며, 이때 요청 정보는 scope 객체에 담겨 전달됩니다. (HTTP의 request랑 유사합니다.)
- scope["query_string"]에는 WebSocket 연결 URL의 쿼리 문자열이 바이트 형태로 포함됩니다.
- 본 구현에서는 ?token=<JWT> 형태로 전달된 JWT를 파싱하여 추출합니다.
해당 방식을 선택한 이유:
- 브라우저 WebSocket API는 Authorization 헤더 설정을 지원하지 않습니다.
- 헤더로 전달하는 방식이 아닌 웹 환경에서 JWT를 전달할 수 있는 현실적인 방법 중 하나가 쿼리 스트링이었습니다.

1-2. JWT 토큰 검증 및 사용자 식별자 추출

@database_sync_to_async
def get_user_from_token(self, token):
    try:
        access_token = AccessToken(token)
        return access_token["user_id"]
    except (TokenError, InvalidToken) as e:
        print(f"토큰 검증 실패: {e}")
        return None

설명:
- simplejwt의 AccessToken을 사용해 JWT의 서명 및 만료 여부를 검증합니다.
- 토큰이 유효한 경우 payload에 포함된 user_id만 추출하여 반환합니다.
- 토큰이 만료되었거나 위·변조된 경우 예외를 발생시키고 None을 반환합니다.
- Django ORM 및 JWT 검증 로직은 동기 코드이므로 database_sync_to_async를 사용해 이벤트 루프 블로킹을 방지했습니다.
해당 방식의 특징:

1-3. WebSocket 연결 시점에 인증 처리

async def __call__(self, scope, receive, send):
    token = self.get_token_from_scope(scope)

    if token is not None:
        user_id = await self.get_user_from_token(token)
        if user_id:
            scope["user_id"] = user_id
        else:
            scope["error"] = "invalid_token"
    else:
        scope["error"] = "no_token"

    return await super().__call__(scope, receive, send)

설명:
- 해당 메서드는 WebSocket 연결 요청마다 가장 먼저 실행되는 진입 지점입니다.
- WebSocket 연결 요청에서 get_token_from_scope()로 JWT 토큰 추출합니다.
- 토큰이 존재하면 get_user_from_token()을 통해 JWT 검증을 수행합니다.
  - 검증에 성공하면 scope[“user_id”]에 사용자 식별자 저장합니다.
  - 토큰이 없거나 검증에 실패한 경우, 오류 상태를 scope에 기록합니다.
- 이후 Consumer 로직으로 요청을 전달하게 됩니다.
해당 구현의 특징:
- 미들웨어에서는 JWT의 진위 여부까지만 판단하고 인증 결과만 scope에 기록하는 역할만 수행합니다.
- 실제 연결 허용 여부 (accept() 호출)는 Consumer에서 처리하도록 책임을 분리했다는 점입니다.
- 이를 통해, WebSocket 연결 단위의 인증 정보를 Consumer 전반에서 재사용할 수 있으며 인증 정책이 바뀌어도 미들웨어 단에서 확장 및 변경이 가능해집니다.

1-4. 전체코드

# chatbot/middleware.py

from urllib.parse import parse_qs

from channels.db import database_sync_to_async
from channels.middleware import BaseMiddleware
from rest_framework_simplejwt.exceptions import InvalidToken, TokenError
from rest_framework_simplejwt.tokens import AccessToken


class JWTAuthMiddleware(BaseMiddleware):
    """
    Websocket 연결을 위한 JWT 인증 미들웨어

    해당 미들웨어는 Websocket 연결 요청 시 쿼리 문자열에서 JWT 토큰을 추출하고,
    해당 토큰을 검증해서 user_id를 `scope`에 저장합니다.
    인증이 실패하면 `scope`에 어류 메시지를 추가합니다.

    사용 방식
    - 클라이언트는 Websocket 연결 시 `?token=<access_token>` 형태로 토큰을 전달해야 합니다.
    - 유효한 토큰이면 `scope["user_id"]`에 user_id를 추가합니다.
    = 유효하지 않거나 토큰이 없으면 `scope["error"]`에 오류 메시지를 추가합니다.

    매서드(Method)
    - get_token_from_scope(scope): 쿼리 문자열에서 JWT 토큰을 추출하는 매서드
    - get_user_from_token(token): 추출한 토큰을 검증하고, user_id를 반환하는 매서드
    """

    async def __call__(self, scope, receive, send):
        token = self.get_token_from_scope(scope)

        if token is not None:
            user_id = await self.get_user_from_token(token)
            if user_id:
                scope["user_id"] = user_id
            else:
                scope["error"] = "invalid_token"
        else:
            scope["error"] = "no_token"

        return await super().__call__(scope, receive, send)

    def get_token_from_scope(self, scope):
        query_string = parse_qs(scope.get("query_string", b"").decode("utf-8"))
        return query_string.get("token", [None])[0]

    @database_sync_to_async
    def get_user_from_token(self, token):
        try:
            access_token = AccessToken(token)
            return access_token["user_id"]
        except (TokenError, InvalidToken) as e:
            print(f"토큰 검증 실패: {e}")
            return None

2. chatbot/consumers.py 수정

2-1. 사용자 인증 및 조회 기준 정의

@sync_to_async
def get_user(self):
    if self.user_id:
        return User.objects.filter(id=self.user_id).first()
    return None

설명:
- Consumer에서 인증 여부를 판단하는 기준은 “해당 사용자가 실제 존재하는 지 여부”입니다.
- JWT payload만 신뢰하지 않고 요청 시점에 DB 기준으로 검증을 합니다.
- 해당 메서드는 이후 모든 인증 판단의 기준이 됩니다.
- @sync_to_async를 사용해 동기 ORM 호출로 인한 이벤트 루프 블로킹을 방지했습니다.

2-2. WebSocket 연결 시점 인증 처리 (`connect`)

# chatbot/consumers.py

async def connect(self):
    self.user_id = self.scope.get("user_id")
    self.is_authenticated = await self.get_user() is not None

    if not self.is_authenticated:
        await self.close()
        return

    await self.accept()

설명:
- JWTAuthMiddleware에서 검증된 user_id를 scope에서 가져옵니다.
- get_user() 매서드를 통해 실제 사용자 존재 여부를 다시 확인합니다.
- 인증에 실패한 경우 accept()를 호출하지 않고 즉시 연결을 종료합니다.
이 방식의 의도:
- WebSocket 인증을 연결 단위에서 강제하기 위함
- 인증되지 않은 사용자가 WebSocket 채널에 진입하는 것 자체를 차단
- 인증 판단을 Consumer에서 명시적으로 통제

2-3. 메시지 수신 시 인증 상태 재검증 (`receive`)

async def receive(self, text_data):
        if not self.is_authenticated:
            await self.send(
                text_data=json.dumps(
                    {"error": "인증되지 않은 사용자입니다."},
                    ensure_ascii=False,
                )
            )

        data = json.loads(text_data)
        user_message = data["message"]

        llm_response = await self.llm_response(user_message)
        await self.send(
            text_data=json.dumps(
                {"response": llm_response},
                ensure_ascii=False,
            )
        )
    async def llm_response(self, user_message):
        llm_response = get_chatbot_response(user_message)
        response = llm_response.get("response")
        return response

설명:
- 연결이 이미 성립된 이후에도 인증 상태를 다시 확인합니다.
- 인증되지 않은 상태에서 메시지가 전달되는 것을 방어적으로 차단합니다.
이 로직을 추가한 이유:
- WebSocket은 연결이 장시간 유지되므로 상태를 무조건 신뢰하지 않기 위함
- 예기치 않은 상태 변화나 비정상적인 흐름에 대비한 defensive programming
- 인증 로직을 “한 번의 검사”가 아니라 지속적으로 유지해야 할 상태로 다루기 위함

2-4. 전체코드

import json

from asgiref.sync import sync_to_async
from channels.generic.websocket import AsyncWebsocketConsumer

from accounts.models import User

from .utils import get_chatbot_response


class ChatConsumer(AsyncWebsocketConsumer):
    """
    WebSocket을 통한 실시간 채팅을 처리하는 comsumer 클래스
    해당 클래스는 Django Channels의 AsyncWebsocketConsumer를 상속받아
    WebSocket 연결을 관리하고, 사용자 인증 및 AI응답을 처리합니다.
    사용 방식:
    - WebSocket 연결 시 사용자를 인증하고, 인증된 사용자만 채팅 이용 가능
    - 클라이언트가 메시지를 전송하면 Chatbot 모델이 응답을 반환
    - 인증되지 않은 사용자는 메시지를 전송할 수 없으며, 에러 메시지를 반환
    매서드(Method)
    - connect(): WebSocket 연결을 초기화하고 사용자 인증을 수행함.
    - disconnect(close_code): Websocket 연결 종료
    - receive(text_data): 클라이언트의 메시지를 받아 Chatbot에 전달
    - llm_response(user_message): Chatbot에서 응답을 받아옴
    - get_user(): user_id를 기반으로 데이터베이스에서 사용자 정보 조회
    """

    async def connect(self):
        self.user_id = self.scope.get("user_id")
        self.is_authenticated = await self.get_user() is not None

        if not self.is_authenticated:
            await self.close()
            return

        await self.accept()

    async def disconnect(self, close_code):
        pass

    async def receive(self, text_data):
        if not self.is_authenticated:
            await self.send(
                text_data=json.dumps(
                    {"error": "인증되지 않은 사용자입니다."},
                    ensure_ascii=False,
                )
            )

        data = json.loads(text_data)
        user_message = data["message"]

        llm_response = await self.llm_response(user_message)
        await self.send(
            text_data=json.dumps(
                {"response": llm_response},
                ensure_ascii=False,
            )
        )
    async def llm_response(self, user_message):
        llm_response = get_chatbot_response(user_message)
        response = llm_response.get("response")
        return response

    @sync_to_async
    def get_user(self):
        if self.user_id:
            return User.objects.filter(id=self.user_id).first()
        return None

3. config/asgi.py 수정

3-1. `AuthMiddlewareStack` → `JWTAuthMiddleware` 변경

import os

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

from chatbot.middleware import JWTAuthMiddleware
from chatbot.routing import websocket_urlpatterns

os.environ.setdefault("DJANGO_SETTINGS_MODULE", "config.settings")

application = ProtocolTypeRouter(
    {
        "http": get_asgi_application(),
        "websocket": JWTAuthMiddleware(URLRouter(websocket_urlpatterns)),
    }
)

설명:
- 기존의 WebSocket 요청에 사용하던 AuthMiddlewareStack(URLRouter(websocket_urlpatterns)을 JWTAuthMiddleware(URLRouter(websocket_urlpatterns)로 변경하였습니다.
- AuthMiddlewareStack는 Django의 세션(Session) 기반 인증을 전제로 동작합니다.
  - 내부적으로는 쿠키에 저장된 세션 정보를 기반으로 scope[“user”]를 구성합니다.
- AINFO는 세션을 사용하지 않고 JWT 기반 인증을 사용하고 있기 때문에, AuthMiddlewareStack를 그대로 사용할 경우 WebSocket 환경에서는 인증 정보가 전달되지 않습니다.
- 따라서, WebSocket 연결 시 JWT를 직접 검증할 수 있도록 커스텀으로 구현한 JWTAuthMiddleware를 적용하도록 구성했습니다.

🔄 마무리

구현 성과

이번 구현은 기본적으로 제공되는 AuthMiddlewareStack이 세션 기반 인증을 전제로 동작한다는 것을 알게 되었고, WebSocket 연결 시점(handshake)에 JWT를 직접 검증하는 커스텀 미들웨어를 구현함으로써 WebSocket 구간에서도 인증되지 않은 사용자의 접근을 차단할 수 있는 구조를 마련했습니다.

회고

이번 프로젝트를 진행하면서 팀 내에서 미리 약속했던 것이 “AI 도구(GPT, Cursor 등)을 사용할 수 있는 있지만, 코드를 그대로 작성해 달라는 식의 사용은 지양하고 최대한 직접 고민하고 이해하는 방향으로 학습하자”였습니다.

Django Channels 환경에서 JWT 인증을 적용하는 자료는 생각보다 많지 않았습니다. 프로젝트를 진행하던 시점(2025년 2월)에는 한국어로 정리된 글은 거의 찾아볼 수 없었고, 해외 자료도 몇 개 찾을 수는 있었지만 그대로 적용하기에는 맞지 않는 부분이 많았습니다. 그리하여 있는 자료를 참고해서 전체적인 흐름을 이해한 뒤, 우리 프로젝트에 상황에 맞제 하나씩 수정하면서 구현했었습니다.

그 과정에서 왜 다른 사람들의 코드를 그대로 적용할 수 없는지, 우리 프로젝트랑 무엇이 다르기 때문에 어떤 부분을 수정해야 하는지에 대해 계속 고민하고 공부했던 경험이 개인적으로 많은 도움이 되었고 무엇보다 즐거운 학습과정이었습니다.

또한, WebSocket JWT 인증을 구현하고 해당 내용을 정리해 노션에 기술 문서 형태로 공유했었는데, 프로젝트 중간에 다른 팀의 팀원이 해당 문서를 보고 WebSocket JWT 인증을 구현할 수 있었고, 실제로 많은 도움이 되었다는 이야기를 들었습니다. 제가 정리한 내용이 누군가의 구현에 실제로 도움이 되었다는 것은 제가 해당 내용을 학습하고 고민하던 노력이 의미 있었으며, 저 뿐만 아니라 다같이 성장하고 발전할 수 있었다는 것이 뿌듯하고 좋았습니다.

Project / Research	Period	Description
AINFO: Personalized Public Service Recommendation Chatbot	Mar 2025 – Apr 2025	LLM-based chatbot for personalized public service recommendations. Built with Django/DRF and deployed prototype.
H2O2-induced astrocytic collagen triggers neuronal death via fucosylation-dependent glial barrier formation upon ischemic stroke	Jan 2013 – Nov 2014	Performed RNA-seq/ATAC-seq data analysis using Python, R, and Linux, covering gene expression analysis with DESeq2/edgeR and visualization with ggplot2

📌 관련 개념 정리

🧭 배경

🧩 구현 과정

1. chatbot/middleware.py 추가

1-1. WebSocket 연결 요청에서 JWT 토큰 추출

1-2. JWT 토큰 검증 및 사용자 식별자 추출

1-3. WebSocket 연결 시점에 인증 처리

1-4. 전체코드

2. chatbot/consumers.py 수정

2-1. 사용자 인증 및 조회 기준 정의

2-2. WebSocket 연결 시점 인증 처리 (connect)

2-3. 메시지 수신 시 인증 상태 재검증 (receive)

2-4. 전체코드

3. config/asgi.py 수정

3-1. AuthMiddlewareStack → JWTAuthMiddleware 변경

🔄 마무리

구현 성과

회고

Templates (for web app):

Error

2-2. WebSocket 연결 시점 인증 처리 (`connect`)

2-3. 메시지 수신 시 인증 상태 재검증 (`receive`)

3-1. `AuthMiddlewareStack` → `JWTAuthMiddleware` 변경