websocket

WebSocket セキュリティスキル

ファイル構成

• SKILL.md：中核となる原則、パターン、基本的なセキュリティ（このファイル）
• references/security-examples.md：CSWSH の例と認証パターン
• references/advanced-patterns.md：接続管理、スケーリングパターン
• references/threat-model.md：CSWSH を含む攻撃シナリオ

検証ゲート

ゲート 0.2：合格（5 件以上の脆弱性を記録） - CVE-2024-23898、CVE-2024-26135、CVE-2023-0957

---

1. 概要

リスクレベル：高

理由：WebSocket 接続は Same-Origin Policy による保護を回避するため、クロスサイト WebSocket ハイジャック（CSWSH）の脆弱性があります。永続的な接続には、慎重な認証、セッション管理、入力検証が必要です。

あなたは WebSocket セキュリティの専門家であり、双方向の永続的な接続に特有の脆弱性を理解しています。

中核となる専門分野

• CSWSH（クロスサイト WebSocket ハイジャック）の防止
• Origin ヘッダーの検証とトークンベースの認証
• メッセージの検証とメッセージごとの認可
• レート制限と接続ライフサイクルのセキュリティ

---

2. 中核となる責務

基本原則

1. TDD を最優先：実装前にテストを作成し、セキュリティ境界と接続ライフサイクルをテストします
2. パフォーマンスを考慮：低レイテンシ（<50ms）、接続プーリング、バックプレッシャーに最適化します
3. Origin を検証：Origin ヘッダーが明示的な許可リストに含まれているか、必ず確認します
4. 最初に認証：メッセージを受け付ける前に、本人確認を行います
5. アクションごとに認可：接続が無制限のアクセスを意味するとは想定しません
6. すべてのメッセージを検証：WebSocket メッセージを信頼できない入力として扱います
7. リソースを制限：メッセージにレート制限を設け、アイドル状態の接続をタイムアウトさせます

セキュリティ判断フレームワーク

状況	対応
新しい接続	Origin を検証し、認証トークンを必須にします
各メッセージ	形式を検証し、アクションの認可を確認します
機密性の高い操作	セッションを再検証し、操作を記録します
アイドル状態の接続	一定時間操作がなければタイムアウトさせます
エラー発生時	接続を閉じ、詳細を記録します

---

3. 技術基盤

推奨バージョン

コンポーネント	バージョン	備考
FastAPI/Starlette	0.115+	WebSocket をサポート
websockets	12.0+	Python 用 WebSocket ライブラリ

セキュリティ設定

WEBSOCKET_CONFIG = {
    "max_message_size": 1024 * 1024,  # 1MB
    "max_connections_per_ip": 10,
    "idle_timeout_seconds": 300,
    "messages_per_minute": 60,
}

# NEVER use "*" for origins
ALLOWED_ORIGINS = ["https://app.example.com", "https://admin.example.com"]

---

4. 実装ワークフロー（TDD）

ステップ 1：まず失敗するテストを作成

import pytest
from httpx import AsyncClient, ASGITransport
from fastapi.testclient import TestClient

# Test security boundaries first
@pytest.mark.asyncio
async def test_origin_validation_rejects_invalid():
    """CSWSH prevention - must reject invalid origins."""
    async with AsyncClient(
        transport=ASGITransport(app=app),
        base_url="http://test"
    ) as client:
        # This should fail until origin validation is implemented
        with pytest.raises(Exception):
            async with client.websocket_connect(
                "/ws?token=valid",
                headers={"Origin": "https://evil.com"}
            ):
                pass

@pytest.mark.asyncio
async def test_authentication_required():
    """Must reject connections without valid token."""
    with TestClient(app) as client:
        with pytest.raises(Exception):
            with client.websocket_connect("/ws") as ws:
                pass

@pytest.mark.asyncio
async def test_message_authorization():
    """Each message action must be authorized."""
    with TestClient(app) as client:
        with client.websocket_connect(
            "/ws?token=readonly_user",
            headers={"Origin": "https://app.example.com"}
        ) as ws:
            ws.send_json({"action": "delete", "id": "123"})
            response = ws.receive_json()
            assert response.get("error") == "Permission denied"

ステップ 2：テストに合格する最小限の実装

# Implement only what's needed to pass the test
async def validate_origin(websocket: WebSocket) -> bool:
    origin = websocket.headers.get("origin")
    if not origin or origin not in ALLOWED_ORIGINS:
        await websocket.close(code=4003, reason="Invalid origin")
        return False
    return True

ステップ 3：リファクタリングして検証

# Run all WebSocket tests
pytest tests/websocket/ -v --asyncio-mode=auto

# Check for security issues
bandit -r src/websocket/

# Verify no regressions
pytest tests/ -v

---

5. パフォーマンスパターン

パターン 1：接続プーリング

# BAD - Create new connection for each request
ws = await create_connection(user_id)  # Expensive!

# GOOD - Reuse connections from pool
class ConnectionPool:
    def __init__(self, max_size: int = 100):
        self.connections: dict[str, WebSocket] = {}

    async def get_or_create(self, user_id: str) -> WebSocket:
        if user_id not in self.connections:
            self.connections[user_id] = await create_connection(user_id)
        return self.connections[user_id]

パターン 2：メッセージのバッチ処理

# BAD - Send messages one at a time
for item in items:
    await websocket.send_json({"type": "item", "data": item})

# GOOD - Batch messages to reduce overhead
await websocket.send_json({"type": "batch", "data": items[:50]})

パターン 3：バイナリプロトコル

# BAD - JSON for high-frequency data (~80 bytes)
await websocket.send_json({"x": 123.456, "y": 789.012, "z": 456.789})

# GOOD - Binary format (20 bytes)
import struct
await websocket.send_bytes(struct.pack('!3f', 123.456, 789.012, 456.789))

パターン 4：ハートビートの最適化

# BAD - Fixed frequent heartbeats
HEARTBEAT_INTERVAL = 5  # Every 5 seconds

# GOOD - Adaptive heartbeats based on activity
interval = 60 if (time() - last_activity) < 60 else 30

パターン 5：バックプレッシャーの処理

# BAD - Blocks on slow clients
await ws.send_json(message)

# GOOD - Timeout and bounded queue
from collections import deque
queue = deque(maxlen=100)  # Drop oldest when full
try:
    await asyncio.wait_for(ws.send_json(message), timeout=1.0)
except asyncio.TimeoutError:
    pass  # Client too slow

---

6. 実装パターン

パターン 1：Origin の検証（CSWSH 防止に不可欠）

from fastapi import WebSocket

async def validate_origin(websocket: WebSocket) -> bool:
    """Validate WebSocket origin against allowlist."""
    origin = websocket.headers.get("origin")
    if not origin or origin not in ALLOWED_ORIGINS:
        await websocket.close(code=4003, reason="Invalid origin")
        return False
    return True

@app.websocket("/ws")
async def websocket_endpoint(websocket: WebSocket):
    if not await validate_origin(websocket):
        return
    await websocket.accept()

パターン 2：トークンベースの認証

from jose import jwt, JWTError

async def authenticate_websocket(websocket: WebSocket) -> User | None:
    """Authenticate via token (not cookies - vulnerable to CSWSH)."""
    token = websocket.query_params.get("token")
    if not token:
        await websocket.close(code=4001, reason="Authentication required")
        return None
    try:
        payload = jwt.decode(token, SECRET_KEY, algorithms=["HS256"])
        user = await user_service.get(payload.get("sub"))
        if not user:
            await websocket.close(code=4001, reason="User not found")
            return None
        return user
    except JWTError:
        await websocket.close(code=4001, reason="Invalid token")
        return None

パターン 3：メッセージごとの認可

from pydantic import BaseModel, field_validator

class WebSocketMessage(BaseModel):
    action: str
    data: dict

    @field_validator('action')
    @classmethod
    def validate_action(cls, v):
        if v not in {'subscribe', 'unsubscribe', 'send', 'query'}:
            raise ValueError(f'Invalid action: {v}')
        return v

async def handle_message(websocket: WebSocket, user: User, raw_data: dict):
    try:
        message = WebSocketMessage(**raw_data)
    except ValueError:
        await websocket.send_json({"error": "Invalid message format"})
        return

    if not user.has_permission(f"ws:{message.action}"):
        await websocket.send_json({"error": "Permission denied"})
        return

    result = await handlers[message.action](user, message.data)
    await websocket.send_json(result)

パターン 4：レート制限付き接続マネージャー

from collections import defaultdict
from time import time

class SecureConnectionManager:
    def __init__(self):
        self.connections: dict[str, WebSocket] = {}
        self.message_counts: dict[str, list[float]] = defaultdict(list)
        self.connections_per_ip: dict[str, int] = defaultdict(int)

    async def connect(self, websocket: WebSocket, user_id: str, ip: str) -> bool:
        if self.connections_per_ip[ip] >= WEBSOCKET_CONFIG["max_connections_per_ip"]:
            await websocket.close(code=4029, reason="Too many connections")
            return False
        await websocket.accept()
        self.connections[user_id] = websocket
        self.connections_per_ip[ip] += 1
        return True

    def check_rate_limit(self, user_id: str) -> bool:
        now = time()
        self.message_counts[user_id] = [
            ts for ts in self.message_counts[user_id] if ts > now - 60
        ]
        if len(self.message_counts[user_id]) >= WEBSOCKET_CONFIG["messages_per_minute"]:
            return False
        self.message_counts[user_id].append(now)
        return True

パターン 5：セキュアなハンドラーの全体例

@app.websocket("/ws")
async def websocket_endpoint(websocket: WebSocket):
    if not await validate_origin(websocket):
        return
    user = await authenticate_websocket(websocket)
    if not user:
        return

    ip = websocket.client.host
    if not await manager.connect(websocket, user.id, ip):
        return

    try:
        while True:
            raw = await asyncio.wait_for(
                websocket.receive_json(),
                timeout=WEBSOCKET_CONFIG["idle_timeout_seconds"]
            )
            if not manager.check_rate_limit(user.id):
                await websocket.send_json({"error": "Rate limited"})
                continue
            await handle_message(websocket, user, raw)
    except (WebSocketDisconnect, asyncio.TimeoutError):
        pass
    finally:
        manager.disconnect(user.id, ip)

---

7. セキュリティ標準

ドメインの脆弱性状況

CVE ID	重大度	説明	緩和策
CVE-2024-23898	高	Jenkins の CSWSH - コマンド実行	Origin を検証
CVE-2024-26135	高	MeshCentral の CSWSH - 設定漏えい	Origin + SameSite
CVE-2023-0957	重大	Gitpod の CSWSH - アカウント乗っ取り	Origin + トークン認証

OWASP Top 10 との対応

カテゴリー	緩和策
A01 アクセス制御	Origin の検証、メッセージごとの認可
A02 暗号化の失敗	TLS/WSS のみを使用、署名付きトークン
A03 インジェクション	すべてのメッセージ内容を検証
A07 認証の失敗	トークン認証、セッション検証

CSWSH 防止のまとめ

async def secure_websocket_handler(websocket: WebSocket):
    # 1. VALIDATE ORIGIN (Critical)
    if websocket.headers.get("origin") not in ALLOWED_ORIGINS:
        await websocket.close(code=4003)
        return
    # 2. AUTHENTICATE with token (not cookies)
    user = await validate_token(websocket.query_params.get("token"))
    if not user:
        await websocket.close(code=4001)
        return
    # 3. Accept only after validation
    await websocket.accept()
    # 4. AUTHORIZE each message, 5. RATE LIMIT, 6. TIMEOUT idle

---

8. よくある間違いとアンチパターン

オリジン検証なし

# NEVER - vulnerable to CSWSH
@app.websocket("/ws")
async def vulnerable(websocket: WebSocket):
    await websocket.accept()  # Accepts any origin!

# ALWAYS - validate origin first
if websocket.headers.get("origin") not in ALLOWED_ORIGINS:
    await websocket.close(code=4003)
    return

Cookie のみの認証

# NEVER - cookies sent automatically in CSWSH attacks
session = websocket.cookies.get("session")

# ALWAYS - require explicit token parameter
token = websocket.query_params.get("token")

メッセージごとの認可なし

# NEVER - assumes connection = full access
if data["action"] == "delete":
    await delete_resource(data["id"])

# ALWAYS - check permission for each action
if not user.has_permission("delete"):
    return {"error": "Permission denied"}

入力検証なし

# NEVER - trust WebSocket messages
await db.execute(f"SELECT * FROM {data['table']}")  # SQL injection!

# ALWAYS - validate with Pydantic
message = WebSocketMessage(**data)

---

9. 実装前チェックリスト

フェーズ 1：コードを書く前

• セキュリティ境界（CSWSH、認証、認可）に関する失敗するテストを書く
• 接続ライフサイクル（接続、切断、タイムアウト）に関する失敗するテストを書く
• メッセージ検証とレート制限に関する失敗するテストを書く
• references/threat-model.md の脅威モデルを確認する
• パフォーマンス要件（レイテンシー、スループット、接続数）を特定する

フェーズ 2：実装中

• 明示的な許可リストに対してオリジンを検証する
• トークンベースの認証を使用する（Cookie のみにはしない）
• メッセージごとに認可を確認する
• レート制限とアイドルタイムアウトを実装する
• すべてのメッセージを Pydantic で検証する
• 効率化のためにコネクションプーリングを行う
• 低速なクライアントに対するバックプレッシャー処理を実装する

フェーズ 3：コミット前

• すべてのセキュリティテストに合格する：pytest tests/websocket/ -v
• セキュリティ上の問題がないことを確認する：bandit -r src/websocket/
• 本番環境の設定で WSS（TLS）を強制する
• CSWSH のテストカバレッジを確認する
• パフォーマンスベンチマークが目標を満たす（レイテンシー <50ms）
• リグレッションがないことを確認する：pytest tests/ -v

---

10. まとめ

セキュリティ目標：

• CSWSH 耐性：オリジン検証、トークン認証
• 適切な認可：メッセージごとの権限確認
• レート制限：メッセージの大量送信を防止
• 検証済み：すべてのメッセージを信頼できないものとして扱う

重要な注意事項：必ずオリジンを検証し、トークン認証を使用し（Cookie は使用せず）、各メッセージを認可し、本番環境では WSS を使用してください。