Posted inPython

ScapyによるPCAP解析：パケットの「覗き見」からポートスキャンの自動検知まで

Python tutorial - IT technology blog
Python tutorial - IT technology blog

なぜWiresharkではなくScapyを使うのか？

ISPのネットワーク監視システムに半年間携わった経験から、ある厳しい現実に気づきました。Wiresharkは単発のバグ調査には最適ですが、1日に500個のPCAPファイルを処理したり、即時アラートが必要な場合には限界があります。そこで力を発揮するのがPython + Scapyです。

Scapyは単なるファイル読み込みライブラリではありません。L2からL4までの各データビットに直接干渉できます。私のプロジェクトも、最初はIPフィルタリングのためのわずか200行のコードから始まりましたが、振る舞い検知モジュールを追加した後は2000行にまで成長しました。ここでは、独学で何週間も費やすことなく、このツールを使いこなすためのロードマップを紹介します。

クイックスタート：5分でPCAPファイルを読み込む

まず、システムをクリーンに保つために、仮想環境（venv）にScapyをインストールしましょう。

pip install scapy

任意の capture.pcap ファイルで試してみましょう。中身を探索するための最も短いコードがこちらです：

from scapy.all import rdpcap

# PCAPファイルを読み込む
packets = rdpcap('capture.pcap')

# 最初の10パケットの概要を表示
for packet in packets[:10]:
    print(packet.summary())

重要な注意点：rdpcap 関数は全データをRAMにロードします。1GB程度のPCAPファイルの場合、8GB RAMのノートPCでは即座にフリーズする可能性があります。大容量ファイルの処理方法については、記事の最後で解説します。

パケットの解析：プロトコルレイヤーを深掘りする

ScapyはTCP/IPモデルのようにパケットをレイヤー（layers）ごとに階層化します。packet[Layer] という構文を使って、非常に素早くデータにアクセスできます。

IPおよびTCP/UDPの抽出

解析作業の大部分は、送信元IP、宛先IP、およびポート番号を中心に展開されます。以下のコードは、これらの情報を素早くフィルタリングするのに役立ちます：

from scapy.all import rdpcap, IP, TCP

pkts = rdpcap('capture.pcap')
for pkt in pkts:
    if pkt.haslayer(IP):
        src_ip = pkt[IP].src
        dst_ip = pkt[IP].dst
        
        if pkt.haslayer(TCP):
            src_port = pkt[TCP].sport
            dst_port = pkt[TCP].dport
            print(f"[{src_ip}:{src_port}] -> [{dst_ip}:{dst_port}]")

DNSおよびHTTPデータの探索

DNS情報を取得するには DNSQR レイヤーを確認する必要があります。HTTPの場合は、手動パースの時間を節約するために scapy.layers.http モジュールを活用しましょう。

from scapy.all import *
from scapy.layers.http import HTTPRequest

def process_packet(pkt):
    if pkt.haslayer(DNSQR):
        query = pkt[DNSQR].qname.decode()
        print(f"[DNS] クエリ: {query}")

    if pkt.haslayer(HTTPRequest):
        url = pkt[HTTPRequest].Host.decode() + pkt[HTTPRequest].Path.decode()
        print(f"[HTTP] リクエスト: {url}")

# sniffを使用してパケットを1つずつ処理し、RAM使用量を最小限に抑える
sniff(offline='capture.pcap', prn=process_packet, store=0)

実践：ポートスキャン検知スクリプトの作成

ポートスキャンは、泥棒が隙を探して各部屋のドアを叩いて回るようなものです。攻撃者は通常、大量のTCP SYNパケットをさまざまなポートに送信します。

実装ロジック：特定の送信元IPが1つの宛先の100以上のポートに対してSYNパケットを送信した場合、警告フラグを立てます。

from scapy.all import rdpcap, IP, TCP
from collections import defaultdict

def detect_port_scan(pcap_file, threshold=100):
    packets = rdpcap(pcap_file)
    scan_attempts = defaultdict(set)

    for pkt in packets:
        # TCP SYNフラグ(0x02)を持つパケットを確認
        if pkt.haslayer(IP) and pkt.haslayer(TCP) and pkt[TCP].flags == 0x02:
            src = pkt[IP].src
            dst = pkt[IP].dst
            dport = pkt[TCP].dport
            scan_attempts[(src, dst)].add(dport)

    for (src, dst), ports in scan_attempts.items():
        if len(ports) > threshold:
            print(f"[!] 警告: {src} が {dst} の {len(ports)} 個のポートをスキャンしています")

detect_port_scan('capture.pcap')

大容量データ処理における実戦経験

実際の現場で扱うPCAPファイルは、ラボの例題よりもはるかに過酷です。私が得た3つの秘訣を紹介します：

  • rdpcapは忘れましょう： 大容量ファイルには sniff(offline='file.pcap', store=0) を使用します。store=0 パラメータにより、Scapyは処理済みのパケットをメモリに保持しないため、RAM使用量を安定させることができます。
  • BPFフィルタの活用： Webトラフィックのみが必要な場合は、全パケットを読み込まないでください。sniff関数内で filter="tcp port 80 or 443" を指定します。これによりScapyがカーネル階層でデータをフィルタリングするため、処理速度が最大60%向上します。
  • ゴミデータへの対策： 実際のネットワークデータはデコードエラーが発生しがちです。.decode() 処理は常に try-except で囲み、数百万行のログの途中でスクリプトが停止しないようにしましょう。

Scapyを使いこなすことで、無限の自動化の可能性が広がります。画面上で一行ずつクリックする代わりに、スクリプトを自動実行してレポートをTelegramに直接送信することも可能です。ぜひ活用してみてください！

Share:
Tags: