⚠️ 免責事項：本記事の内容は自分が管理するアプリ・または許可を得たシステムのみを対象としてください。第三者のシステムへの無断アクセスは不正アクセス禁止法に抵触します。
WebSocketを使ったアプリを開発・デバッグしていると、「python websocket 傍受 プロキシ 実装」がどうしても必要になる場面があります。「どんなデータが流れているのか見えない」「認証トークンが平文で飛んでいないか確認したい」――そんな悩みを抱えているエンジニアは多いはずです。
そこで今回は、Python 3.10以上 + websockets 13.x以降の新しいAPIを使って、ws://・wss://の両方に対応したWebSocketプロキシを一から自作し、全通信をJSON形式でログ保存するところまで実装しながら解説します。セキュリティ検証やペネトレーションテスト学習にも直結する内容ですので、ぜひ手を動かしながら読み進めてください。
この記事で分かること
- WebSocketプロキシのアーキテクチャと asyncio を使う理由
- ws:// と wss:// 両対応のPython傍受プロキシの完全実装コード
- 通信ログをJSON形式で保存し、特定メッセージをフィルタ・改ざんする方法
WebSocket傍受が必要になる場面
「WebSocketの通信内容を確認したい」というニーズは、開発現場で意外と頻繁に発生します。HTTPと違い、WebSocketはブラウザのDevToolsだけでは深い解析がしにくく、独自プロキシが最も確実な手段になることが多いです。
デバッグ：何が流れているかを可視化する
フロントエンドとバックエンドがWebSocketで通信するアプリを開発中、「サーバーから想定外のメッセージが来ている気がする」という場面は珍しくありません。python websocket 傍受 プロキシ 実装を自前で用意すれば、すべての送受信メッセージをコンソールやファイルに記録でき、原因特定が格段に速くなります。
セキュリティ検証：トークンが平文で流れていないか
自社プロダクトのセキュリティ診断では、認証トークンやセッションIDが暗号化なしでWebSocketメッセージに含まれていないかを確認する必要があります。wss://（TLS）でも、暗号化の外側のペイロード設計に問題がある場合があり、プロキシで中身を検査することが重要です。
APIリバースエンジニアリング：自分のアプリの通信仕様を確認
外部SDKや古いコードが出力するWebSocketメッセージの仕様が不明なとき、プロキシを噛ませてペイロードを記録すれば、仕様書を掘り起こさなくてもAPIの通信フォーマットを把握できます。
プロキシの仕組みをアーキテクチャ図で理解する
実装に入る前に、全体像を掴んでおきましょう。WebSocketプロキシは「クライアント」「プロキシ（自作Pythonサーバー）」「本物のサーバー」の3者で構成されます。
3者構成のアスキーアート図
┌──────────┐ ws://localhost:8080 ┌─────────────────────┐ ws://実サーバー ┌────────────────┐
│ クライアント │ ──────────────────────▶ │ Pythonプロキシサーバー │ ─────────────────▶ │ WebSocketサーバー │
│ (ブラウザ等) │ ◀────────────────────── │ (今回自作する部分) │ ◀───────────────── │ (接続先) │
└──────────┘ └─────────────────────┘ └────────────────┘
│
▼
┌─────────────┐
│ ログ出力・保存 │
│ (JSON file) │
└─────────────┘
クライアントはプロキシを「本物のサーバー」だと思って接続します。プロキシは受け取ったメッセージをそのまま（あるいは改ざんして）本物のサーバーへ転送し、逆方向も同様に中継します。この構造を中間プロキシ（Man-in-the-Middle Proxy）と呼びます。
asyncioを使う理由
WebSocketプロキシの核心は「クライアント→サーバー」と「サーバー→クライアント」の同時双方向転送です。スレッドを使っても実装できますが、asyncioのノンブロッキングI/O（I/O待ちの間に別の処理を実行できる仕組み）を使うと、シングルプロセスで何百ものWebSocket接続を効率よく捌けます。websockets 13.0以降の新API（websockets.asyncio）はこのasyncioを前提とした設計になっています。
ws://（非暗号化）プロキシを実装する
まずはTLSなしの ws:// で動く基本プロキシを作ります。シンプルな構成で動作原理を理解してから、次のセクションでTLS対応に拡張します。
websocketsライブラリのインストール
websockets 13.0以降では websockets.asyncio 名前空間が標準となり、旧来の websockets.legacy は非推奨になっています。必ず最新版をインストールしてください。
# websocketsの最新版をインストール（13.x以降が対象）
pip install "websockets>=13.0"
双方向転送プロキシの完全実装コード
プロキシの心臓部は asyncio.gather() による2方向の同時転送です。片方が切断されたらもう片方も閉じる、というライフサイクル管理も忘れずに入れます。
# ファイル名: ws_proxy.py
# Python 3.10以上対象 | websockets 13.x以降の新API使用
import asyncio
import json
import datetime
import sys
from websockets.asyncio.server import serve
from websockets.asyncio.client import connect
from websockets.exceptions import ConnectionClosed
# ── 設定 ──────────────────────────────────────────
LISTEN_HOST = "localhost"
LISTEN_PORT = 8080
TARGET_URI = "ws://localhost:9000" # 転送先の本物のサーバーURI
# ──────────────────────────────────────────────────
def log_message(direction: str, payload: str | bytes) -> None:
"""送受信メッセージをコンソールに色付きで出力する"""
arrow = "▶ CLIENT→SERVER" if direction == "send" else "◀ SERVER→CLIENT"
ts = datetime.datetime.now().isoformat(timespec="milliseconds")
# バイナリは16進数で表示
if isinstance(payload, bytes):
display = f"[binary {len(payload)} bytes] {payload[:64].hex()}..."
else:
display = payload[:200] # 長すぎる場合は先頭200文字のみ
print(f"[{ts}] {arrow}\n {display}\n")
async def forward(src, dst, direction: str) -> None:
"""一方向の転送ループ：srcから受信してdstへ転送する"""
try:
async for message in src:
log_message(direction, message)
await dst.send(message)
except ConnectionClosed:
# 接続が閉じられたら正常終了
pass
except Exception as e:
print(f"[ERROR] forward({direction}): {e}", file=sys.stderr)
async def proxy_handler(client_ws) -> None:
"""クライアント接続を受け取り、ターゲットサーバーへ中継する"""
client_addr = client_ws.remote_address
print(f"[INFO] 接続受付: {client_addr} → {TARGET_URI}")
try:
# ターゲットサーバーへ接続
async with connect(TARGET_URI) as server_ws:
# クライアント→サーバー と サーバー→クライアント を同時実行
await asyncio.gather(
forward(client_ws, server_ws, "send"),
forward(server_ws, client_ws, "recv"),
)
except OSError as e:
print(f"[ERROR] ターゲットへの接続失敗: {e}", file=sys.stderr)
except Exception as e:
print(f"[ERROR] proxy_handler: {e}", file=sys.stderr)
finally:
print(f"[INFO] 接続終了: {client_addr}")
async def main() -> None:
"""プロキシサーバーを起動するエントリーポイント"""
print(f"[START] WebSocketプロキシ起動: ws://{LISTEN_HOST}:{LISTEN_PORT}")
print(f"[START] 転送先: {TARGET_URI}")
async with serve(proxy_handler, LISTEN_HOST, LISTEN_PORT) as server:
await server.serve_forever()
if __name__ == "__main__":
asyncio.run(main())
動作確認：テスト用エコーサーバーで試す
プロキシ単体でテストするには、まず転送先となるシンプルなエコーサーバーを立てます。別ターミナルでエコーサーバーを起動し、さらに別のターミナルでプロキシを起動してください。
# ファイル名: echo_server.py（テスト用：本番環境では不要）
import asyncio
from websockets.asyncio.server import serve
async def echo(websocket):
# 受信したメッセージをそのまま返すだけのシンプルなサーバー
async for message in websocket:
await websocket.send(f"ECHO: {message}")
async def main():
async with serve(echo, "localhost", 9000) as server:
print("[TEST SERVER] ws://localhost:9000 で起動中")
await server.serve_forever()
if __name__ == "__main__":
asyncio.run(main())
# ターミナル1：エコーサーバー起動
python echo_server.py
# ターミナル2：プロキシ起動
python ws_proxy.py
# ターミナル3：websocketsの対話クライアントで動作確認
python -m websockets ws://localhost:8080
wss://（TLS暗号化）に対応させる
実際の本番WebSocketは wss://（WebSocket over TLS）を使うため、TLS対応は必須です。ここでは ssl.SSLContext を使ってプロキシ両端のTLS設定を行います。
ssl.SSLContextの設定コード
プロキシがTLSを扱う際は「クライアントからプロキシへのTLS」と「プロキシからサーバーへのTLS」の2つのコンテキストが必要です。クライアント側には自己署名証明書を使い、サーバー側は標準のCA検証を行います。自己署名証明書の作成方法については、「PythonとOpenSSLで自己署名SSL証明書を自動生成する完全ガイド」を参照してください。
# ファイル名: wss_proxy.py
# TLS対応版WebSocketプロキシ（wss://対応）
import asyncio
import ssl
import sys
from websockets.asyncio.server import serve
from websockets.asyncio.client import connect
from websockets.exceptions import ConnectionClosed
# ── 設定 ──────────────────────────────────────────
LISTEN_HOST = "localhost"
LISTEN_PORT = 8443
TARGET_URI = "wss://echo.websocket.org" # TLS対応の転送先URI
# 自己署名証明書のパス（生成方法は前記事を参照）
CERT_FILE = "server.crt"
KEY_FILE = "server.key"
# ──────────────────────────────────────────────────
def create_server_ssl_context() -> ssl.SSLContext:
"""クライアント→プロキシ間のTLSコンテキスト（自己署名証明書を使用）"""
ctx = ssl.SSLContext(ssl.PROTOCOL_TLS_SERVER)
try:
ctx.load_cert_chain(certfile=CERT_FILE, keyfile=KEY_FILE)
except FileNotFoundError:
print("[ERROR] 証明書ファイルが見つかりません。先に証明書を生成してください。")
raise
return ctx
def create_client_ssl_context(verify: bool = True) -> ssl.SSLContext:
"""プロキシ→実サーバー間のTLSコンテキスト"""
if verify:
# 本番：システムのCA証明書でサーバー証明書を検証
ctx = ssl.create_default_context()
else:
# 開発・検証用：証明書検証をスキップ（本番では使わないこと）
ctx = ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT)
ctx.check_hostname = False
ctx.verify_mode = ssl.CERT_NONE
return ctx
def log_message(direction: str, payload: str | bytes) -> None:
import datetime
arrow = "▶ C→S" if direction == "send" else "◀ S→C"
ts = datetime.datetime.now().isoformat(timespec="milliseconds")
if isinstance(payload, bytes):
display = f"[binary {len(payload)}B]"
else:
display = payload[:200]
print(f"[{ts}] {arrow} {display}")
async def forward(src, dst, direction: str) -> None:
"""一方向の転送ループ"""
try:
async for message in src:
log_message(direction, message)
await dst.send(message)
except ConnectionClosed:
pass
except Exception as e:
print(f"[ERROR] forward: {e}", file=sys.stderr)
async def proxy_handler(client_ws) -> None:
"""wss://クライアントを受け取り、TLSでターゲットへ中継する"""
client_ssl = create_client_ssl_context(verify=True)
try:
async with connect(TARGET_URI, ssl=client_ssl) as server_ws:
await asyncio.gather(
forward(client_ws, server_ws, "send"),
forward(server_ws, client_ws, "recv"),
)
except Exception as e:
print(f"[ERROR] proxy_handler: {e}", file=sys.stderr)
async def main() -> None:
server_ssl = create_server_ssl_context()
print(f"[START] TLS対応プロキシ起動: wss://{LISTEN_HOST}:{LISTEN_PORT}")
async with serve(
proxy_handler,
LISTEN_HOST,
LISTEN_PORT,
ssl=server_ssl,
) as server:
await server.serve_forever()
if __name__ == "__main__":
asyncio.run(main())
証明書エラーの回避方法
クライアント（ブラウザやPythonスクリプト）がプロキシの自己署名証明書を信頼しない場合、接続エラーになります。自己署名証明書の作成・システムへのインポート手順は「PythonとOpenSSLで自己署名SSL証明書を自動生成する完全ガイド」で詳しく解説しています。テスト用Pythonクライアントなら以下のように証明書検証をスキップできます。
# テスト用クライアント：自己署名証明書のプロキシに接続する例
import asyncio
import ssl
from websockets.asyncio.client import connect
async def test_client():
# 証明書検証をスキップ（テスト・検証目的のみ）
ssl_ctx = ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT)
ssl_ctx.check_hostname = False
ssl_ctx.verify_mode = ssl.CERT_NONE
async with connect("wss://localhost:8443", ssl=ssl_ctx) as ws:
await ws.send("Hello, TLS Proxy!")
response = await ws.recv()
print(f"受信: {response}")
asyncio.run(test_client())
ログをJSON形式でファイルに保存する
コンソール出力だけでは後で解析しにくいため、各メッセージを構造化されたJSONとしてファイルに保存します。タイムスタンプ・方向・ペイロードを1行1JSONで記録する「JSON Lines（JSONL）」形式が、後処理しやすくておすすめです。
構造化JSONログの実装コード
以下のコードは先ほどの ws_proxy.py に差し込む形で使えます。log_message() 関数を置き換えてください。
# ファイル名: logger.py（ws_proxy.pyからimportして使う）
import json
import datetime
import threading
from pathlib import Path
LOG_FILE = Path("ws_proxy_log.jsonl")
# ファイル書き込みをスレッドセーフにするためのロック
_lock = threading.Lock()
def log_message(direction: str, payload: str | bytes, session_id: str = "") -> None:
"""
メッセージをJSON Lines形式でファイルに追記する。
direction: "send"（クライアント→サーバー）または "recv"（サーバー→クライアント）
"""
# バイナリペイロードはbase64エンコードして保存
if isinstance(payload, bytes):
import base64
payload_data = base64.b64encode(payload).decode()
is_binary = True
else:
payload_data = payload
is_binary = False
entry = {
"timestamp": datetime.datetime.now(datetime.timezone.utc).isoformat(),
"session_id": session_id,
"direction": direction, # "send" or "recv"
"is_binary": is_binary,
"size_bytes": len(payload),
"payload": payload_data,
}
line = json.dumps(entry, ensure_ascii=False)
# ファイルへのスレッドセーフな追記
with _lock:
with LOG_FILE.open("a", encoding="utf-8") as f:
f.write(line + "\n")
# コンソールにも出力
arrow = "▶ C→S" if direction == "send" else "◀ S→C"
print(f"[{entry['timestamp']}] {arrow} ({entry['size_bytes']}B) "
f"{str(payload_data)[:100]}")
ws_proxy.py の先頭に from logger import log_message を追加して、既存の log_message 定義を削除すれば完成です。
ログファイルをリアルタイムでtailする方法
ログが蓄積されていく様子をリアルタイムで確認したい場合、Linuxなら tail -f コマンドが使えます。Windowsや自動化が必要な場面ではPythonで同等の処理を実装できます。
# Linux/macOS：ログをリアルタイムで確認する
tail -f ws_proxy_log.jsonl | python -c "
import sys, json
for line in sys.stdin:
try:
d = json.loads(line)
print(f\"{d['timestamp']} [{d['direction']}] {d['payload'][:120]}\")
except Exception:
pass
"
# ファイル名: tail_log.py（Windows・クロスプラットフォーム対応版）
import time
import json
from pathlib import Path
LOG_FILE = Path("ws_proxy_log.jsonl")
def tail_log(poll_interval: float = 0.2) -> None:
"""ログファイルを継続的に監視し、新しい行をリアルタイム出力する"""
print(f"[TAIL] {LOG_FILE} を監視中... (Ctrl+C で停止)")
with LOG_FILE.open("r", encoding="utf-8") as f:
# ファイル末尾にシーク
f.seek(0, 2)
while True:
line = f.readline()
if not line:
time.sleep(poll_interval)
continue
try:
entry = json.loads(line)
arrow = "▶" if entry["direction"] == "send" else "◀"
print(f"{entry['timestamp']} {arrow} {entry['payload'][:120]}")
except json.JSONDecodeError:
pass
if __name__ == "__main__":
try:
tail_log()
except KeyboardInterrupt:
print("\n[TAIL] 監視を停止しました。")
フィルタリングと応用：特定メッセージを改ざんする
プロキシの応用として、特定のキーワードを含むメッセージのみを抽出したり、内容を書き換えて転送したりする機能を追加します。セキュリティ検証の文脈では、「正規のメッセージを改ざんするとサーバーはどう応答するか？」をテストする際に使われます。
特定キーワードを含むメッセージのみ抽出するフィルター
forward() 関数の中でフィルター処理を行います。条件に合致しないメッセージは通常通り転送し、マッチしたメッセージだけログに詳細記録するパターンが実用的です。
# ファイル名: proxy_with_filter.py
# フィルタリング機能付きWebSocketプロキシ（ws_proxy.pyの拡張版）
import asyncio
import json
import datetime
import re
import sys
from websockets.asyncio.server import serve
from websockets.asyncio.client import connect
from websockets.exceptions import ConnectionClosed
LISTEN_HOST = "localhost"
LISTEN_PORT = 8080
TARGET_URI = "ws://localhost:9000"
# フィルター対象のキーワードリスト（正規表現で指定可能）
FILTER_PATTERNS = [
r"token", # "token" を含むメッセージ
r"auth", # "auth" を含むメッセージ
r"password", # "password" を含むメッセージ
]
def should_alert(payload: str | bytes) -> bool:
"""フィルターパターンにマッチするか判定する"""
if isinstance(payload, bytes):
try:
text = payload.decode("utf-8", errors="replace")
except Exception:
return False
else:
text = payload
for pattern in FILTER_PATTERNS:
if re.search(pattern, text, re.IGNORECASE):
return True
return False
def log_alert(direction: str, payload: str | bytes, matched: bool) -> None:
ts = datetime.datetime.now().isoformat(timespec="milliseconds")
arrow = "▶ C→S" if direction == "send" else "◀ S→C"
mark = "🚨 [ALERT]" if matched else " "
display = payload[:200] if isinstance(payload, str) else payload[:64].hex()
print(f"[{ts}] {mark} {arrow} {display}")
async def forward(src, dst, direction: str) -> None:
"""フィルター付き一方向転送ループ"""
try:
async for message in src:
matched = should_alert(message)
log_alert(direction, message, matched)
if matched:
# アラート対象のメッセージはJSON形式で詳細ログに保存
entry = {
"timestamp": datetime.datetime.now(datetime.timezone.utc).isoformat(),
"direction": direction,
"payload": message if isinstance(message, str) else message.hex(),
"alert": True,
}
with open("alert_log.jsonl", "a", encoding="utf-8") as f:
f.write(json.dumps(entry, ensure_ascii=False) + "\n")
# マッチしてもしなくても転送は行う（透過プロキシとして動作）
await dst.send(message)
except ConnectionClosed:
pass
except Exception as e:
print(f"[ERROR] forward({direction}): {e}", file=sys.stderr)
async def proxy_handler(client_ws) -> None:
try:
async with connect(TARGET_URI) as server_ws:
await asyncio.gather(
forward(client_ws, server_ws, "send"),
forward(server_ws, client_ws, "recv"),
)
except Exception as e:
print(f"[ERROR] proxy_handler: {e}", file=sys.stderr)
async def main() -> None:
print(f"[START] フィルター付きプロキシ: ws://{LISTEN_HOST}:{LISTEN_PORT}")
async with serve(proxy_handler, LISTEN_HOST, LISTEN_PORT) as server:
await server.serve_forever()
if __name__ == "__main__":
asyncio.run(main())
メッセージ内容を書き換えて転送する改ざん実証コード（検証目的）
テスト対象のサーバーが入力バリデーションを適切に行っているかを確認するため、プロキシ側でメッセージを書き換えてから転送する実装です。自分が管理するシステムの検証目的のみに使用してください。
# ファイル名: proxy_tamper.py
# メッセージ改ざん機能付きプロキシ（セキュリティ検証専用）
import asyncio
import json
import sys
from websockets.asyncio.server import serve
from websockets.asyncio.client import connect
from websockets.exceptions import ConnectionClosed
LISTEN_HOST = "localhost"
LISTEN_PORT = 8080
TARGET_URI = "ws://localhost:9000"
def tamper_message(direction: str, payload: str | bytes) -> str | bytes:
"""
メッセージを改ざんして返す関数。
この関数を編集して検証したい改ざんパターンを設定する。
"""
if direction != "send":
# サーバー→クライアント方向は改ざんしない
return payload
if isinstance(payload, bytes):
return payload # バイナリは今回はスキップ
# JSON形式のメッセージを改ざんする例
try:
data = json.loads(payload)
# 例：roleフィールドを "admin" に書き換えてサーバーの反応を確認
if "role" in data:
original = data["role"]
data["role"] = "admin"
print(f"[TAMPER] role: {original!r} → 'admin'")
return json.dumps(data, ensure_ascii=False)
# 例：amountフィールドを0に書き換えてバリデーションを確認
if "amount" in data:
original = data["amount"]
data["amount"] = 0
print(f"[TAMPER] amount: {original!r} → 0")
return json.dumps(data, ensure_ascii=False)
except (json.JSONDecodeError, KeyError):
pass # JSONでないメッセージはそのまま転送
return payload
async def forward(src, dst, direction: str) -> None:
"""改ざん処理を組み込んだ転送ループ"""
try:
async for message in src:
modified = tamper_message(direction, message)
if modified != message:
print(f" 元メッセージ : {str(message)[:100]}")
print(f" 改ざん後 : {str(modified)[:100]}")
await dst.send(modified)
except ConnectionClosed:
pass
except Exception as e:
print(f"[ERROR] {e}", file=sys.stderr)
async def proxy_handler(client_ws) -> None:
try:
async with connect(TARGET_URI) as server_ws:
await asyncio.gather(
forward(client_ws, server_ws, "send"),
forward(server_ws, client_ws, "recv"),
)
except Exception as e:
print(f"[ERROR] proxy_handler: {e}", file=sys.stderr)
async def main() -> None:
print(f"[START] 改ざんプロキシ起動: ws://{LISTEN_HOST}:{LISTEN_PORT}")
print("[WARNING] このツールは自分が管理するシステムの検証のみに使用すること")
async with serve(proxy_handler, LISTEN_HOST, LISTEN_PORT) as server:
await server.serve_forever()
if __name__ == "__main__":
asyncio.run(main())
まとめ・FAQ・次の記事への導線
まとめ
- python websocket 傍受 プロキシ 実装は、websockets 13.x以降の
websockets.asyncioAPIとasyncio.gather()を使った双方向転送が基本構造 - ws://（非暗号化）は
serve()+connect()だけで実装でき、wss://（TLS）はssl.SSLContextを両端に設定するだけで対応可能 - JSON Lines形式のログ保存により、後からjqなどで高速な解析・集計が行える
- フィルター機能で認証トークンや機密データが平文で流れていないかを自動検出できる
- 改ざんプロキシでサーバー側のバリデーション実装を実際のリクエストで検証できる
よくある質問（FAQ）
Q. mitmproxyとの違いは何ですか？
A. mitmproxyはHTTP/HTTPS全般を対象とした多機能ツールで、WebSocketも扱えますがGUI・CLI両対応の汎用ツールです。本記事の自作プロキシはWebSocket専用でコードが完全に手元にあるため、フィルタリングや改ざんロジックを自由にカスタマイズできます。また、mitmproxyはPython拡張（アドオン）を書く必要がある部分も、今回の実装では直接Pythonコードに組み込めるのが利点です。
Q. Dockerコンテナ内でも使えますか？
A. はい。コンテナ内でプロキシを動かす場合、LISTEN_HOST を "0.0.0.0" に変更し、コンテナのポートマッピング（例：-p 8080:8080）を設定してください。コンテナ間通信でプロキシを噛ませるには、docker-composeのネットワーク設定で TARGET_URI をコンテナ名（例：ws://app-server:9000）に変えるだけで動作します。
Q. WebSocketの接続が途中で切れてしまいます。どうすればいいですか？
A. Pingフレームによるキープアライブが原因のことが多いです。websocketsはデフォルトで20秒ごとにPingを送りますが、プロキシがPingを正しく中継しないと接続が切れます。今回の実装は生メッセージのみを転送するため、本番利用では connect(ping_interval=None) でプロキシ側のPingを無効化し、クライアント・サーバー間のPingをそのまま透過させる設計を検討してください。
次に読むべき記事
- 「PyJWT完全ガイド：署名・検証・改ざん検知・有効期限まで実装で理解する」
- 「PythonとOpenSSLで自己署名SSL証明書を自動生成する完全ガイド」
- 「Pythonで学ぶネットワークパケット解析：scapyで通信を可視化する」