如何在SvelteKit中处理带认证的WebSocket连接

方案1：通过SvelteKit API端点中转获取Token #

核心思路：利用SvelteKit服务端端点可读取Http-Only Cookie的特性，创建一个专门的API接口提取JWT Token，客户端调用该接口拿到Token后再建立WebSocket连接。

步骤1：创建服务端API端点 #

新建src/routes/api/get-websocket-token/+server.ts：

import { error, json } from '@sveltejs/kit';
import type { RequestHandler } from './$types';

export const GET: RequestHandler = async ({ cookies, locals }) => {
  // 从Http-Only Cookie中读取access token（替换成你的Cookie名）
  const accessToken = cookies.get('access_token');
  
  if (!accessToken) {
    throw error(401, '未登录');
  }

  // 可选：通过locals验证用户身份（需提前在hooks.server.ts中解析Cookie并注入locals）
  if (!locals.user) {
    throw error(401, '无效的登录状态');
  }

  return json({ accessToken });
};

步骤2：客户端调用接口并建立连接 #

在页面组件（如src/routes/+page.svelte）中实现WebSocket连接逻辑：

<script lang="ts">
  let socket: WebSocket | null = null;

  async function connectWebSocket() {
    try {
      // 调用API获取Token
      const res = await fetch('/api/get-websocket-token');
      if (!res.ok) throw new Error('获取Token失败');
      const { accessToken } = await res.json();

      // 构造带Token的WebSocket URL（替换成你的Django后端地址）
      const wsUrl = `wss://your-django-backend.com/ws/endpoint/?token=${encodeURIComponent(accessToken)}`;
      socket = new WebSocket(wsUrl);

      socket.onopen = () => console.log('WebSocket连接成功');
      socket.onmessage = (event) => console.log('收到消息:', event.data);
      socket.onerror = (err) => console.error('WebSocket错误:', err);
      socket.onclose = () => {
        console.log('WebSocket连接关闭');
        socket = null;
      };
    } catch (err) {
      console.error('连接失败:', err);
    }
  }

  // 页面加载时自动连接，或绑定到按钮点击事件
  connectWebSocket();
</script>

<button on:click={connectWebSocket} disabled={!!socket}>
  {socket ? '已连接' : '连接WebSocket'}
</button>

安全提示 #

• 该API端点默认仅允许同域访问，可在hooks.server.ts中添加额外校验（如校验请求来源）
• 不要将Token存储到localStorage或全局变量，用完即丢弃
• 确保Cookie设置了Secure、Http-Only、SameSite=Strict属性

方案2：页面加载时从服务端注入Token #

核心思路：在页面的服务端加载函数中读取Cookie里的Token，将其作为页面props传递给客户端，直接用于WebSocket连接。

步骤1：服务端加载函数传递Token #

修改src/routes/+page.server.ts：

import type { PageServerLoad } from './$types';
import { error } from '@sveltejs/kit';

export const load: PageServerLoad = async ({ cookies, locals }) => {
  const accessToken = cookies.get('access_token');
  if (!accessToken || !locals.user) {
    throw error(401, '请先登录');
  }
  return { accessToken };
};

步骤2：客户端接收并使用Token #

在页面组件中接收props并建立连接：

<script lang="ts">
  import type { PageData } from './$types';
  export let data: PageData;

  let socket: WebSocket | null = null;

  function connectWebSocket() {
    const wsUrl = `wss://your-django-backend.com/ws/endpoint/?token=${encodeURIComponent(data.accessToken)}`;
    socket = new WebSocket(wsUrl);

    // 后续的onopen/onmessage等逻辑同方案1
  }

  connectWebSocket();
</script>

安全提示 #

• 此方案会将Token暴露给客户端，存在XSS风险，必须确保页面无XSS漏洞
• 设置严格的内容安全策略（CSP），禁止内联脚本和未授权的脚本来源

方案3：修改Django后端支持Cookie认证WebSocket（推荐，若允许修改后端） #

核心思路：让Django的WebSocket中间件直接从Cookie中读取JWT Token，无需客户端手动传递，浏览器会自动带上Http-Only Cookie。

步骤1：自定义Channels认证中间件 #

在Django项目中新建middleware.py：

from channels.middleware import BaseMiddleware
from channels.db import database_sync_to_async
from rest_framework_simplejwt.authentication import JWTAuthentication
from rest_framework.exceptions import AuthenticationFailed

class JWTCookieAuthMiddleware(BaseMiddleware):
    async def __call__(self, scope, receive, send):
        # 从WebSocket请求的Cookie中获取Token
        cookies = scope.get('cookies', {})
        access_token = cookies.get('access_token')
        if not access_token:
            raise AuthenticationFailed('未提供认证Token')
        
        # 用SimpleJWT解析Token（需同步转异步）
        auth = JWTAuthentication()
        try:
            user, token = await database_sync_to_async(auth.authenticate_credentials)(access_token)
            scope['user'] = user
        except AuthenticationFailed as e:
            raise AuthenticationFailed(str(e))
        
        return await super().__call__(scope, receive, send)

步骤2：配置Channels路由 #

修改asgi.py：

import os
from django.core.asgi import get_asgi_application
from channels.routing import ProtocolTypeRouter, URLRouter
from .middleware import JWTCookieAuthMiddleware
import your_app.routing  # 替换成你的WebSocket路由模块

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

django_asgi_app = get_asgi_application()

application = ProtocolTypeRouter({
    "http": django_asgi_app,
    "websocket": JWTCookieAuthMiddleware(
        URLRouter(
            your_app.routing.websocket_urlpatterns
        )
    ),
})

步骤3：客户端简化连接 #

此时客户端无需处理Token，直接连接即可：

<script lang="ts">
  let socket: WebSocket | null = null;

  function connectWebSocket() {
    socket = new WebSocket('wss://your-django-backend.com/ws/endpoint/');
    // 后续逻辑同前
  }
</script>

方案选择建议 #

• 若无法修改Django后端：优先选择方案1（API中转），安全性高于方案2
• 若可以修改后端：优先选择方案3，Token全程不暴露给客户端，安全性最高

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