如何让你的WebSocket连接既安全又高效？

2025年7月10日
| 没有评论


扫描二维码
关注或者微信搜一搜：编程智域 前端至全栈交流与成长

发现1000+提升效率与开发的AI工具和实用程序：https://tools.cmdragon.cn/

第五章：安全防护与性能优化

5.1 WebSocket安全认证（JWT集成）

实现原理：通过HTTP头携带JWT令牌完成握手阶段的认证，认证失败时返回403状态码中断连接。需使用websockets
库处理连接状态，结合jwt库进行令牌解码。

认证流程图解：

[客户端] --携带JWT--> [握手请求]
[服务端] --解码令牌--> [验证权限]
验证成功 --> 建立WebSocket连接
验证失败 --> 返回HTTP 403

sequenceDiagram participant 客户端 participant 握手请求 participant 服务端 客户端 ->> 握手请求: 携带JWT 握手请求 ->> 服务端: 转发JWT 服务端 ->> 服务端: 解码令牌 服务端 ->> 服务端: 验证权限 alt 验证成功 服务端 ->> 客户端: 建立WebSocket连接 else 验证失败 服务端 ->> 客户端: 返回HTTP 403 end

核心代码实现（需要安装依赖：python-jose[cryptography]==3.3.0）：

from fastapi import WebSocket, HTTPException
from jose import JWTError, jwt


async def websocket_auth(websocket: WebSocket):
    token = websocket.headers.get("Authorization")
    if not token:
        raise HTTPException(status_code=403, detail="未携带令牌")
    try:
        payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
        return payload["sub"]
    except JWTError:
        await websocket.close(code=4001)  # 自定义关闭代码


@app.websocket("/ws")
async def websocket_endpoint(websocket: WebSocket):
    await websocket.accept()
    username = await websocket_auth(websocket)
    # 后续业务逻辑

5.2 跨域配置（CORS for WebSocket）

配置要点：需在中间件中显式声明websocket协议，并设置allow_origins白名单：

from fastapi.middleware.cors import CORSMiddleware

app.add_middleware(
    CORSMiddleware,
    allow_origins=["https://your-domain.com"],
    allow_methods=["GET", "POST", "websocket"],
    allow_headers=["Authorization"]
)

5.3 消息频率限制与防DDOS攻击

三层防御策略：

  1. 应用层限流：使用slowapi实现速率限制
from slowapi import Limiter
from slowapi.util import get_remote_address

limiter = Limiter(key_func=get_remote_address)
app.state.limiter = limiter


@app.websocket("/chat")
@limiter.limit("10/minute")  # 每分钟10条消息
async def chat_handler(websocket: WebSocket):
    ...
  1. Nginx限流配置：
limit_req_zone $binary_remote_addr zone=wslimit:10m rate=20r/s;

location /ws {
    limit_req zone=wslimit burst=30;
    proxy_pass http://backend;
    proxy_http_version 1.1;
    proxy_set_header Upgrade $http_upgrade;
    proxy_set_header Connection "upgrade";
}
  1. 连接数熔断：监控活跃连接数
MAX_CONNECTIONS = 1000
active_connections = set()


@app.on_event("startup")
async def startup():
    # 连接数监控任务
    async def monitor():
        while True:
            if len(active_connections) > MAX_CONNECTIONS * 0.9:
                logger.warning("连接数接近上限")
            await asyncio.sleep(10)

    asyncio.create_task(monitor())

5.4 异步处理优化

连接池调优参数（uvicorn启动命令）：

uvicorn main:app --workers 4 --ws websockets --limit-concurrency 2000

异步任务分流示例：

import concurrent.filters


@app.websocket("/data-stream")
async def data_stream(websocket: WebSocket):
    await websocket.accept()
    # CPU密集型任务转线程池执行
    loop = asyncio.get_event_loop()
    result = await loop.run_in_executor(
        None,  # 使用默认线程池
        cpu_intensive_task,
        request_data
    )
    await websocket.send_json(result)

课后 Quiz

问题1：当WebSocket连接因JWT过期被拒绝时，客户端应该如何处理？
解析：客户端需要捕获连接关闭事件（code 4001），触发令牌刷新流程，获取新令牌后重新建立连接。

问题2：如何验证CORS配置是否对WebSocket生效？
解析：可通过浏览器开发者工具的Network面板查看握手请求的Response Headers中是否包含Access-Control-Allow-Origin字段。

常见报错解决

报错1：403 Forbidden during WebSocket handshake
原因：未正确传递Authorization头或CORS配置未包含websocket协议
解决：

  1. 检查前端连接代码是否设置headers
new WebSocket(`ws://api.com/ws`, {
    headers: {Authorization: `Bearer ${token}`}
})
  1. 确认服务端CORS中间件的allow_methods包含websocket

报错2：RuntimeError: Cannot call "receive" once a close message has been sent.
原因：在连接关闭后仍尝试操作WebSocket对象
解决：在所有的await websocket.receive()调用处添加try-except块：

try:
    data = await websocket.receive_json()
except WebSocketDisconnect:
    break  # 退出消息循环

余下文章内容请点击跳转至 个人博客页面 或者 扫码关注或者微信搜一搜：编程智域 前端至全栈交流与成长
，阅读完整的文章：如何让你的WebSocket连接既安全又高效？

往期文章归档：

免费好用的热门在线工具

flask, python, pytorch, tornado