プロトコルリバースエンジニアリング

ネットワークプロトコルをキャプチャ、分析、ドキュメント化するための包括的なテクニック。セキュリティ研究、相互運用性、デバッグに使用できます。

トラフィックキャプチャ

Wireshark キャプチャ

# 特定のインターフェースでキャプチャ
wireshark -i eth0 -k

# フィルター付きでキャプチャ
wireshark -i eth0 -k -f "port 443"

# ファイルにキャプチャ
tshark -i eth0 -w capture.pcap

# リングバッファキャプチャ（ファイルをローテーション）
tshark -i eth0 -b filesize:100000 -b files:10 -w capture.pcap

tcpdump キャプチャ

# 基本的なキャプチャ
tcpdump -i eth0 -w capture.pcap

# フィルター付き
tcpdump -i eth0 port 8080 -w capture.pcap

# 特定のバイト数をキャプチャ
tcpdump -i eth0 -s 0 -w capture.pcap  # フルパケット

# リアルタイム表示
tcpdump -i eth0 -X port 80

中間者攻撃キャプチャ

# HTTP/HTTPS向けmitmproxy
mitmproxy --mode transparent -p 8080

# SSL/TLSインターセプション
mitmproxy --mode transparent --ssl-insecure

# ファイルにダンプ
mitmdump -w traffic.mitm

# Burp Suite
# ブラウザプロキシを 127.0.0.1:8080 に設定

プロトコル分析

Wireshark 分析

# ディスプレイフィルター
tcp.port == 8080
http.request.method == "POST"
ip.addr == 192.168.1.1
tcp.flags.syn == 1 && tcp.flags.ack == 0
frame contains "password"

# ストリーム追跡
右クリック > Follow > TCP Stream
右クリック > Follow > HTTP Stream

# オブジェクトをエクスポート
File > Export Objects > HTTP

# 復号化
Edit > Preferences > Protocols > TLS
  - (Pre)-Master-Secret log filename
  - RSA keys list

tshark 分析

# 特定フィールドを抽出
tshark -r capture.pcap -T fields -e ip.src -e ip.dst -e tcp.port

# 統計情報
tshark -r capture.pcap -q -z conv,tcp
tshark -r capture.pcap -q -z endpoints,ip

# フィルターして抽出
tshark -r capture.pcap -Y "http" -T json > http_traffic.json

# プロトコルの階層構造
tshark -r capture.pcap -q -z io,phs

Scapy による カスタム分析

from scapy.all import *

# pcapを読み込む
packets = rdpcap("capture.pcap")

# パケットを分析
for pkt in packets:
    if pkt.haslayer(TCP):
        print(f"Src: {pkt[IP].src}:{pkt[TCP].sport}")
        print(f"Dst: {pkt[IP].dst}:{pkt[TCP].dport}")
        if pkt.haslayer(Raw):
            print(f"Data: {pkt[Raw].load[:50]}")

# パケットをフィルター
http_packets = [p for p in packets if p.haslayer(TCP)
                and (p[TCP].sport == 80 or p[TCP].dport == 80)]

# カスタムパケットを作成
pkt = IP(dst="target")/TCP(dport=80)/Raw(load="GET / HTTP/1.1\r\n")
send(pkt)

プロトコル識別

一般的なプロトコルシグネチャ

HTTP        - "HTTP/1." または "GET " または "POST " が開始時にある
TLS/SSL     - 0x16 0x03 (レコードレイヤー)
DNS         - UDP ポート 53、特定のヘッダー形式
SMB         - 0xFF 0x53 0x4D 0x42 ("SMB" シグネチャ)
SSH         - "SSH-2.0" バナー
FTP         - "220 " レスポンス、"USER " コマンド
SMTP        - "220 " バナー、"EHLO" コマンド
MySQL       - 0x00 長さプリフィックス、プロトコルバージョン
PostgreSQL  - 0x00 0x00 0x00 スタートアップ長
Redis       - "*" RESP配列プリフィックス
MongoDB     - 特定ヘッダー付きのBSONドキュメント

プロトコルヘッダーパターン

+--------+--------+--------+--------+
|  マジックナンバー / シグネチャ       |
+--------+--------+--------+--------+
|  バージョン    |  フラグ           |
+--------+--------+--------+--------+
|  長さ          |  メッセージタイプ  |
+--------+--------+--------+--------+
|  シーケンス番号 / セッションID      |
+--------+--------+--------+--------+
|  ペイロード...                     |
+--------+--------+--------+--------+

バイナリプロトコル分析

構造識別

# バイナリプロトコルの一般的なパターン

# 長さプリフィックス付きメッセージ
struct Message {
    uint32_t length;      # メッセージ全体の長さ
    uint16_t msg_type;    # メッセージタイプ識別子
    uint8_t  flags;       # フラグ/オプション
    uint8_t  reserved;    # パディング/アライメント
    uint8_t  payload[];   # 可変長ペイロード
};

# Type-Length-Value (TLV)
struct TLV {
    uint8_t  type;        # フィールドタイプ
    uint16_t length;      # フィールド長
    uint8_t  value[];     # フィールドデータ
};

# 固定ヘッダー + 可変ペイロード
struct Packet {
    uint8_t  magic[4];    # "ABCD" シグネチャ
    uint32_t version;
    uint32_t payload_len;
    uint32_t checksum;    # CRC32 など
    uint8_t  payload[];
};

Python プロトコルパーサー

import struct
from dataclasses import dataclass

@dataclass
class MessageHeader:
    magic: bytes
    version: int
    msg_type: int
    length: int

    @classmethod
    def from_bytes(cls, data: bytes):
        magic, version, msg_type, length = struct.unpack(
            ">4sHHI", data[:12]
        )
        return cls(magic, version, msg_type, length)

def parse_messages(data: bytes):
    offset = 0
    messages = []

    while offset < len(data):
        header = MessageHeader.from_bytes(data[offset:])
        payload = data[offset+12:offset+12+header.length]
        messages.append((header, payload))
        offset += 12 + header.length

    return messages

# TLV構造をパース
def parse_tlv(data: bytes):
    fields = []
    offset = 0

    while offset < len(data):
        field_type = data[offset]
        length = struct.unpack(">H", data[offset+1:offset+3])[0]
        value = data[offset+3:offset+3+length]
        fields.append((field_type, value))
        offset += 3 + length

    return fields

ヘックスダンプ分析

def hexdump(data: bytes, width: int = 16):
    """バイナリデータをヘックスダンプ形式でフォーマット。"""
    lines = []
    for i in range(0, len(data), width):
        chunk = data[i:i+width]
        hex_part = ' '.join(f'{b:02x}' for b in chunk)
        ascii_part = ''.join(
            chr(b) if 32 <= b < 127 else '.'
            for b in chunk
        )
        lines.append(f'{i:08x}  {hex_part:<{width*3}}  {ascii_part}')
    return '\n'.join(lines)

# 出力例：
# 00000000  48 54 54 50 2f 31 2e 31  20 32 30 30 20 4f 4b 0d  HTTP/1.1 200 OK.
# 00000010  0a 43 6f 6e 74 65 6e 74  2d 54 79 70 65 3a 20 74  .Content-Type: t

暗号化分析

暗号化の識別

# エントロピー分析 - 高エントロピーは暗号化/圧縮を示唆
import math
from collections import Counter

def entropy(data: bytes) -> float:
    if not data:
        return 0.0
    counter = Counter(data)
    probs = [count / len(data) for count in counter.values()]
    return -sum(p * math.log2(p) for p in probs)

# エントロピー閾値:
# < 6.0: プレーンテキストまたは構造化データの可能性が高い
# 6.0-7.5: 圧縮済みの可能性がある
# > 7.5: 暗号化またはランダムデータの可能性が高い

# 一般的な暗号化インジケーター
# - 高く均一なエントロピー
# - 明白な構造またはパターンがない
# - 長さがブロックサイズ（AESは16バイト）の倍数
# - 開始部分に可能性のあるIV（AES-CBCは16バイト）

TLS分析

# TLSメタデータを抽出
tshark -r capture.pcap -Y "ssl.handshake" \
    -T fields -e ip.src -e ssl.handshake.ciphersuite

# JA3フィンガープリント（クライアント）
tshark -r capture.pcap -Y "ssl.handshake.type == 1" \
    -T fields -e ssl.handshake.ja3

# JA3Sフィンガープリント（サーバー）
tshark -r capture.pcap -Y "ssl.handshake.type == 2" \
    -T fields -e ssl.handshake.ja3s

# 証明書を抽出
tshark -r capture.pcap -Y "ssl.handshake.certificate" \
    -T fields -e x509sat.printableString

復号化アプローチ

# Pre-master secret ログ（ブラウザ）
export SSLKEYLOGFILE=/tmp/keys.log

# Wiresharkを設定
# Edit > Preferences > Protocols > TLS
# (Pre)-Master-Secret log filename: /tmp/keys.log

# 秘密鍵で復号化（利用可能な場合）
# RSA鍵交換にのみ機能
# Edit > Preferences > Protocols > TLS > RSA keys list

カスタムプロトコルドキュメンテーション

プロトコル仕様テンプレート

# プロトコル名仕様

## 概要

プロトコルの目的と設計の簡潔な説明。

## トランスポート

- レイヤー: TCP/UDP
- ポート: XXXX
- 暗号化: TLS 1.2+

## メッセージ形式

### ヘッダー (12バイト)

| オフセット | サイズ | フィールド | 説明                  |
| ------ | ---- | ------- | ----------------------- |
| 0      | 4    | Magic   | 0x50524F54 ("PROT")     |
| 4      | 2    | Version | プロトコルバージョン (1)   |
| 6      | 2    | Type    | メッセージタイプ識別子 |
| 8      | 4    | Length  | ペイロード長（バイト）  |

### メッセージタイプ

| タイプ | 名前      | 説明            |
| ---- | --------- | ---------------------- |
| 0x01 | HELLO     | 接続開始  |
| 0x02 | HELLO_ACK | 接続受け入れ    |
| 0x03 | DATA      | アプリケーションデータ       |
| 0x04 | CLOSE     | 接続終了 |

### タイプ 0x01: HELLO

| オフセット | サイズ | フィールド      | 説明              |
| ------ | ---- | ---------- | ------------------------ |
| 0      | 4    | ClientID   | ユニークなクライアント識別子 |
| 4      | 2    | Flags      | 接続フラグ         |
| 6      | var  | Extensions | TLVエンコードされた拡張   |

## ステートマシン

[INIT] --HELLO--> [WAIT_ACK] --HELLO_ACK--> [CONNECTED] | DATA/DATA | [CLOSED] <--CLOSE--+

## 例

### 接続確立

クライアント -> サーバー: HELLO (ClientID=0x12345678) サーバー -> クライアント: HELLO_ACK (Status=OK) クライアント -> サーバー: DATA (ペイロード)

Wireshark ディセクター (Lua)

-- custom_protocol.lua
local proto = Proto("custom", "Custom Protocol")

-- フィールドを定義
local f_magic = ProtoField.string("custom.magic", "Magic")
local f_version = ProtoField.uint16("custom.version", "Version")
local f_type = ProtoField.uint16("custom.type", "Type")
local f_length = ProtoField.uint32("custom.length", "Length")
local f_payload = ProtoField.bytes("custom.payload", "Payload")

proto.fields = { f_magic, f_version, f_type, f_length, f_payload }

-- メッセージタイプ名
local msg_types = {
    [0x01] = "HELLO",
    [0x02] = "HELLO_ACK",
    [0x03] = "DATA",
    [0x04] = "CLOSE"
}

function proto.dissector(buffer, pinfo, tree)
    pinfo.cols.protocol = "CUSTOM"

    local subtree = tree:add(proto, buffer())

    -- ヘッダーをパース
    subtree:add(f_magic, buffer(0, 4))
    subtree:add(f_version, buffer(4, 2))

    local msg_type = buffer(6, 2):uint()
    subtree:add(f_type, buffer(6, 2)):append_text(
        " (" .. (msg_types[msg_type] or "Unknown") .. ")"
    )

    local length = buffer(8, 4):uint()
    subtree:add(f_length, buffer(8, 4))

    if length > 0 then
        subtree:add(f_payload, buffer(12, length))
    end
end

-- TCPポート向けに登録
local tcp_table = DissectorTable.get("tcp.port")
tcp_table:add(8888, proto)

アクティブテスト

Boofuzz によるファジング

from boofuzz import *

def main():
    session = Session(
        target=Target(
            connection=TCPSocketConnection("target", 8888)
        )
    )

    # プロトコル構造を定義
    s_initialize("HELLO")
    s_static(b"\x50\x52\x4f\x54")  # Magic
    s_word(1, name="version")       # Version
    s_word(0x01, name="type")       # Type (HELLO)
    s_size("payload", length=4)     # Length フィールド
    s_block_start("payload")
    s_dword(0x12345678, name="client_id")
    s_word(0, name="flags")
    s_block_end()

    session.connect(s_get("HELLO"))
    session.fuzz()

if __name__ == "__main__":
    main()

リプレイと変更

from scapy.all import *

# キャプチャトラフィックをリプレイ
packets = rdpcap("capture.pcap")
for pkt in packets:
    if pkt.haslayer(TCP) and pkt[TCP].dport == 8888:
        send(pkt)

# 変更してリプレイ
for pkt in packets:
    if pkt.haslayer(Raw):
        # ペイロードを変更
        original = pkt[Raw].load
        modified = original.replace(b"client", b"CLIENT")
        pkt[Raw].load = modified
        # チェックサムを再計算
        del pkt[IP].chksum
        del pkt[TCP].chksum
        send(pkt)

ベストプラクティス

分析ワークフロー

1. トラフィックをキャプチャ: 複数のセッション、異なるシナリオ
2. 境界を識別: メッセージの開始/終了マーカー
3. 構造をマップ: 固定ヘッダー、可変ペイロード
4. フィールドを識別: 複数サンプルを比較
5. 形式をドキュメント化: 仕様を作成
6. 理解を検証: パーサー/ジェネレーターを実装
7. エッジケースをテスト: ファジング、境界条件

探すべき一般的なパターン

• メッセージ開始時のマジックナンバー/シグネチャ
• 互換性のためのバージョンフィールド
• 長さフィールド（通常は可変データの前）
• メッセージ識別用のタイプ/オペコードフィールド
• 順序付け用のシーケンス番号
• 整合性用のチェックサム/CRC
• タイミング用のタイムスタンプ
• セッション/接続識別子