performing-network-packet-capture-analysis

ネットワークパケットキャプチャの分析

概要

ネットワークパケットキャプチャ（PCAP／PCAPNG ファイル）は、ネットワークアクティビティを示す究極の情報源であり、ホスト間の通信について反論の余地がない証拠を提供します。PCAP ファイルには、ネットワークセグメント上で送信されたすべてのパケットが記録されているため、データの持ち出し、コマンド＆コントロール通信、ラテラルムーブメント、マルウェアの配信、不正アクセスに関するフォレンジック調査に欠かせません。Wireshark は対話型分析の主要ツールであり、tshark は自動処理やスクリプト作成に対応するコマンドライン機能を提供します。最新の PCAPNG 形式は、インターフェースの説明、キャプチャに関するコメント、高精度なタイムスタンプ、パケットごとの注釈など、追加のメタデータに対応しています。

使用する場面

• ネットワークパケットキャプチャの分析を伴うセキュリティ評価を実施する場合
• 関連するセキュリティインシデントへの対応手順に従う場合
• 定期的なセキュリティテストや監査活動を実施する場合
• 実地テストを通じてセキュリティ制御を検証する場合

前提条件

• プロトコルディセクターを備えた Wireshark 4.x
• tshark コマンドラインツール（Wireshark に含まれています）
• キャプチャと基本的なフィルタリング用の tcpdump
• scapy および pyshark ライブラリを備えた Python 3.8 以降
• PCAP ファイル用の十分なディスク容量（数 GB に及ぶ場合があります）

キャプチャ手法

tcpdump

# Capture all traffic on interface eth0
tcpdump -i eth0 -w capture.pcap

# Capture with rotation (100MB files, keep 10)
tcpdump -i eth0 -w capture_%Y%m%d_%H%M%S.pcap -C 100 -W 10

# Capture specific host traffic
tcpdump -i eth0 host 192.168.1.100 -w host_traffic.pcap

# Capture specific port traffic
tcpdump -i eth0 port 443 -w https_traffic.pcap

# Capture with BPF filter for suspicious ports
tcpdump -i eth0 'port 4444 or port 8080 or port 1337' -w suspicious.pcap

Wireshark 表示フィルター

# HTTP traffic
http

# DNS queries
dns

# SMB file transfers
smb2

# Specific IP communication
ip.addr == 192.168.1.100

# Failed TCP connections
tcp.flags.syn == 1 && tcp.flags.ack == 0

# Large data transfers (potential exfiltration)
tcp.len > 1000

# Specific protocol by port
tcp.port == 4444

# TLS handshakes (SNI extraction)
tls.handshake.type == 1

# HTTP POST requests
http.request.method == "POST"

# DNS queries to suspicious TLDs
dns.qry.name contains ".xyz" or dns.qry.name contains ".top"

# Beaconing detection (regular intervals)
frame.time_delta_displayed > 55 && frame.time_delta_displayed < 65

tshark 分析コマンド

# Extract HTTP URLs from capture
tshark -r capture.pcap -Y "http.request" -T fields -e http.host -e http.request.uri

# Extract DNS queries
tshark -r capture.pcap -Y "dns.flags.response == 0" -T fields -e dns.qry.name | sort -u

# Extract file transfers (HTTP objects)
tshark -r capture.pcap --export-objects http,exported_files/

# Extract SMB file transfers
tshark -r capture.pcap --export-objects smb,smb_files/

# Protocol hierarchy statistics
tshark -r capture.pcap -z io,phs

# Conversation statistics
tshark -r capture.pcap -z conv,tcp

# Extract TLS SNI (Server Name Indication)
tshark -r capture.pcap -Y "tls.handshake.type == 1" -T fields -e tls.handshake.extensions_server_name

# Top talkers by bytes
tshark -r capture.pcap -z endpoints,ip -q

# Extract credentials (FTP, HTTP Basic)
tshark -r capture.pcap -Y "ftp.request.command == USER || ftp.request.command == PASS || http.authorization" -T fields -e ftp.request.arg -e http.authorization

Python による PCAP 分析

from scapy.all import rdpcap, IP, TCP, UDP, DNS, DNSQR, Raw
import os
import sys
import json
from collections import defaultdict, Counter
from datetime import datetime


class PCAPForensicAnalyzer:
    """Forensic analysis of PCAP files using Scapy."""

    def __init__(self, pcap_path: str, output_dir: str):
        self.pcap_path = pcap_path
        self.output_dir = output_dir
        os.makedirs(output_dir, exist_ok=True)
        self.packets = rdpcap(pcap_path)

    def get_conversations(self) -> list:
        """Extract unique IP conversations with byte counts."""
        convos = defaultdict(lambda: {"packets": 0, "bytes": 0})
        for pkt in self.packets:
            if IP in pkt:
                key = tuple(sorted([pkt[IP].src, pkt[IP].dst]))
                convos[key]["packets"] += 1
                convos[key]["bytes"] += len(pkt)

        return [
            {"src": k[0], "dst": k[1], "packets": v["packets"], "bytes": v["bytes"]}
            for k, v in sorted(convos.items(), key=lambda x: x[1]["bytes"], reverse=True)
        ]

    def extract_dns_queries(self) -> list:
        """Extract all DNS queries from the capture."""
        queries = []
        for pkt in self.packets:
            if DNS in pkt and pkt[DNS].qr == 0 and DNSQR in pkt:
                queries.append({
                    "query": pkt[DNSQR].qname.decode(errors="replace").rstrip("."),
                    "type": pkt[DNSQR].qtype,
                    "src": pkt[IP].src if IP in pkt else "unknown"
                })
        return queries

    def detect_beaconing(self, threshold_seconds: float = 5.0) -> list:
        """Detect potential beaconing activity based on regular intervals."""
        ip_timestamps = defaultdict(list)
        for pkt in self.packets:
            if IP in pkt and TCP in pkt:
                key = (pkt[IP].src, pkt[IP].dst, pkt[TCP].dport)
                ip_timestamps[key].append(float(pkt.time))

        beacons = []
        for key, times in ip_timestamps.items():
            if len(times) < 5:
                continue
            deltas = [times[i+1] - times[i] for i in range(len(times)-1)]
            if deltas:
                avg_delta = sum(deltas) / len(deltas)
                variance = sum((d - avg_delta) ** 2 for d in deltas) / len(deltas)
                if variance < threshold_seconds and avg_delta > 1:
                    beacons.append({
                        "src": key[0], "dst": key[1], "port": key[2],
                        "avg_interval": round(avg_delta, 2),
                        "variance": round(variance, 4),
                        "connection_count": len(times)
                    })
        return sorted(beacons, key=lambda x: x["variance"])

    def get_protocol_distribution(self) -> dict:
        """Get protocol distribution statistics."""
        protocols = Counter()
        for pkt in self.packets:
            if TCP in pkt:
                protocols[f"TCP/{pkt[TCP].dport}"] += 1
            elif UDP in pkt:
                protocols[f"UDP/{pkt[UDP].dport}"] += 1
        return dict(protocols.most_common(50))

    def generate_report(self) -> str:
        """Generate comprehensive PCAP analysis report."""
        report = {
            "analysis_timestamp": datetime.now().isoformat(),
            "pcap_file": self.pcap_path,
            "total_packets": len(self.packets),
            "conversations": self.get_conversations()[:50],
            "dns_queries": self.extract_dns_queries()[:200],
            "potential_beacons": self.detect_beaconing(),
            "protocol_distribution": self.get_protocol_distribution()
        }

        report_path = os.path.join(self.output_dir, "pcap_forensic_report.json")
        with open(report_path, "w") as f:
            json.dump(report, f, indent=2)

        print(f"[*] Total packets: {report['total_packets']}")
        print(f"[*] Conversations: {len(report['conversations'])}")
        print(f"[*] DNS queries: {len(report['dns_queries'])}")
        print(f"[*] Potential beacons: {len(report['potential_beacons'])}")
        return report_path


def main():
    if len(sys.argv) < 3:
        print("Usage: python process.py <pcap_file> <output_dir>")
        sys.exit(1)
    analyzer = PCAPForensicAnalyzer(sys.argv[1], sys.argv[2])
    analyzer.generate_report()


if __name__ == "__main__":
    main()

参考資料

• Wireshark ドキュメント：https://www.wireshark.org/docs/
• PCAP 分析の習得：https://insanecyber.com/mastering-pcap-review/
• SANS ネットワークフォレンジック：https://www.sans.org/cyber-security-courses/network-forensics/
• 練習用の公開 PCAP：https://www.netresec.com/?page=PcapFiles