午前3時、アラート、そして説明のつかない50の接続
午前3時12分、アラートが鳴った。本番サーバーのアウトバウンド帯域が通常の20Mbpsから290Mbpsへ急上昇したのだ。SSHで接続し、netstat -an | grep ESTABLISHEDを実行すると、そこにあった——存在するはずのないIPアドレスへの50以上の接続が。
あの夜、多くのエンジニアがするように、Wiresharkを開いて数分間のトラフィックをキャプチャし、パターンを見つけようと数千のパケットを手動でクリックして調べた。原因の特定に40分かかった。問題が難しかったからではなく、ツールの選択が間違っていたからだ。
解決策はPySharkを使った150行のPythonスクリプトだった。今では自分が管理するすべての本番ノードで動いている。アラートが発生しても、40分かかっていた調査が2分以内にフォーマットされたレポートとしてターミナルに届く。
PySharkとは何か
PySharkはTSharkのPythonラッパーだ。TSharkはWiresharkのパケット解析を支えるコマンドラインエンジンで、WiresharkのGUIを使っているとき、実際にワイヤーデータを解析しているのはTSharkだ。PySharkはそのエンジンをクリーンなPython APIとして提供する。
生のソケットキャプチャやScapyとの決定的な違いは、PySharkがすでに解析済みのパケットを提供することだ。宛先ポートを探すためにバイト列を手動で解析する必要はない。packet.tcp.dstportでそのまま取得できる。HTTPメソッドはpacket.http.request_method、DNSクエリはpacket.dns.qry_nameだ。Wiresharkの3,000以上のプロトコル解析器すべてがPythonからアクセス可能になる。
実際の経験から言うと、ネットワークインフラを扱うなら習得必須のスキルの一つだ。プログラムでトラフィックを解析できるようになると、インシデントへの事後対応ではなく、問題になる前にパターンを検出できるようになる。
PySharkには2つのキャプチャモードがある：
- LiveCapture — ネットワークインターフェースからリアルタイムでパケットをキャプチャする
- FileCapture — 既存の
.pcapファイルを読み込む。以前にダンプを保存してあるインシデント後の分析に最適
自動トラフィックアナライザーの構築
環境のセットアップ
PySharkはバックエンドとしてTSharkのインストールが必要だ。Ubuntu/Debianの場合：
# TSharkをインストール（スタンドアロン、Wireshark GUIは不要）
sudo apt-get install tshark
# 非rootでのキャプチャを許可 — ユーザーをwiresharkグループに追加
sudo usermod -aG wireshark $USER
newgrp wireshark
# PySharkをインストール
pip install pysharkCentOS/RHELの場合：sudo dnf install wireshark-cli。グループ名はディストリビューションによってwiresharkまたはpcapになる場合がある。
リアルタイムでのパケットキャプチャ
最もシンプルで実用的なキャプチャから始めよう——すべてのTCPトラフィックの送信元/宛先を表示するだけのもの：
import pyshark
capture = pyshark.LiveCapture(interface='eth0', bpf_filter='tcp')
capture.sniff(timeout=60)
for packet in capture:
try:
src = packet.ip.src
dst = packet.ip.dst
dport = packet.tcp.dstport
print(f"{src} -> {dst}:{dport}")
except AttributeError:
pass # IP以外または不正なパケットはスキップbpf_filter='tcp'はBerkeley Packet Filter構文を使用する——tcpdumpと同じ構文だ。キャプチャ時にフィルタリングすることで、トラフィックの多いインターフェースでもメモリ使用量を大幅に削減できる。
接続の集計と異常検知
インシデント対応中に必要なのは生のパケットごとの出力ではない。集計データだ。どの宛先に最も多くのトラフィックが向かっているか？どのポートか？特定のIPが異常に多い接続を受けていないか？
import pyshark
from collections import defaultdict, Counter
import time
def analyze_traffic(interface='eth0', duration=120, threshold=20):
"""
`duration`秒間トラフィックをキャプチャする。
`threshold`を超える接続数を持つ宛先IPにフラグを立てる。
"""
connection_counts = Counter()
bytes_per_ip = defaultdict(int)
port_counts = Counter()
# BPFフィルター：アウトバウンドTCPのみ、プライベートサブネットを除外
bpf = 'tcp and not (dst net 10.0.0.0/8 or dst net 192.168.0.0/16 or dst net 172.16.0.0/12)'
capture = pyshark.LiveCapture(
interface=interface,
bpf_filter=bpf
)
print(f"[*] {interface}で{duration}秒間キャプチャ中...")
capture.sniff(timeout=duration)
for packet in capture:
try:
dst_ip = packet.ip.dst
dst_port = packet.tcp.dstport
pkt_len = int(packet.length)
connection_counts[dst_ip] += 1
bytes_per_ip[dst_ip] += pkt_len
port_counts[dst_port] += 1
except AttributeError:
continue
anomalies = {
ip: count
for ip, count in connection_counts.items()
if count > threshold
}
return connection_counts, bytes_per_ip, port_counts, anomalies人間が読めるレポートの生成
キャプチャウィンドウが閉じたら、午前3時でも実際に読めるデータにフォーマットする：
def generate_report(connection_counts, bytes_per_ip, port_counts, anomalies):
lines = []
lines.append("=" * 60)
lines.append("ネットワークトラフィック解析レポート")
lines.append(f"生成日時: {time.strftime('%Y-%m-%d %H:%M:%S')}")
lines.append("=" * 60)
lines.append("\n[接続数上位10宛先IP]")
for ip, count in connection_counts.most_common(10):
mb = bytes_per_ip[ip] / (1024 * 1024)
lines.append(f" {ip:<20} {count:>6} 接続 {mb:>8.2f} MB")
lines.append("\n[上位10宛先ポート]")
for port, count in port_counts.most_common(10):
lines.append(f" ポート {port:<8} {count:>6} パケット")
if anomalies:
lines.append("\n[!! 異常 — 接続数超過]")
for ip, count in sorted(anomalies.items(), key=lambda x: -x[1]):
lines.append(f" !! {ip} — {count} 接続")
else:
lines.append("\n[OK] しきい値を超える異常なし")
report = "\n".join(lines)
print(report)
fname = f"traffic_report_{time.strftime('%Y%m%d_%H%M%S')}.txt"
with open(fname, 'w') as f:
f.write(report)
print(f"\n[*] レポートを{fname}に保存しました")
return report実行可能な完全スクリプト
CLIインターフェースですべてをつなぎ合わせる：
if __name__ == '__main__':
import argparse
parser = argparse.ArgumentParser(description='PySharkトラフィックアナライザー')
parser.add_argument('--interface', default='eth0')
parser.add_argument('--duration', type=int, default=120)
parser.add_argument('--threshold', type=int, default=20)
args = parser.parse_args()
conn_counts, bytes_ip, port_counts, anomalies = analyze_traffic(
interface=args.interface,
duration=args.duration,
threshold=args.threshold
)
generate_report(conn_counts, bytes_ip, port_counts, anomalies)# eth0での基本的な2分間キャプチャ
sudo python3 traffic_analyzer.py
# より長いキャプチャ、高い異常しきい値
sudo python3 traffic_analyzer.py --interface ens3 --duration 300 --threshold 50保存済みpcapファイルの解析
インシデント中にtcpdumpでpcapをキャプチャしていた場合は、LiveCaptureをFileCaptureに置き換えることで事後に解析できる：
def analyze_pcap(filepath):
# display_filterはWireshark構文を使用 — BPFより表現力が高い
capture = pyshark.FileCapture(
filepath,
display_filter='tcp and not ip.dst == 192.168.0.0/16'
)
for packet in capture:
try:
ts = packet.sniff_time
src = packet.ip.src
dst = packet.ip.dst
port = packet.tcp.dstport
print(f"{ts} | {src} -> {dst}:{port}")
except AttributeError:
continue
capture.close() # FileCaptureは必ず明示的にクローズするdisplay_filterはBPFではなくWiresharkのディスプレイフィルター構文を使用する。'http.request.method == "POST"'や'dns.qry.name contains "evil"'のような記述が可能だ。これがバックエンドとしてのTSharkの真の力だ。
本番環境向けチェックの追加
基本版はほとんどのインシデントに対応できる。本番監視用には以下を追加しよう：
既知の不審なポートへのトラフィックにフラグを立てる：
SUSPICIOUS_PORTS = {'4444', '6666', '6667', '31337', '1337', '9001', '8080'}
def check_suspicious_ports(port_counts):
hits = {p: c for p, c in port_counts.items() if str(p) in SUSPICIOUS_PORTS}
if hits:
print(f"[!! アラート] 不審なポートでトラフィックを検知: {hits}")
return hits継続的なウィンドウ監視 — キャプチャをループで囲み、5分間のウィンドウでキャプチャし、前のウィンドウと比較する。ウィンドウ間で3倍のスパイクが発生した場合は、絶対しきい値よりも明確な異常シグナルとなる。
Webhookアラート — 異常が検知されたとき、SlackやTelegramのWebhookにPOSTする。cronジョブと組み合わせれば、フルのSIEMスタックなしにパッシブな継続監視が実現できる。
ジオロケーションのコンテキスト — geoip2ライブラリはローカルのMaxMindデータベースを使ってIPを国にマッピングする。接続の80%が突然ユーザーの居住国ではない国にルーティングされていることは、接続数単独よりも明確なシグナルだ。
このツールが置き換えるもの（と置き換えないもの）
このツールは自動トリアージ、ベースライン比較、レポート生成を担う——人間がGUIをクリックしても効果がない、ネットワーク調査の繰り返し作業の部分だ。
特定のTCPストリームを追跡したり、ペイロードバイトを検査したり、HTTPセッションを視覚的に再構築したりする必要があるときはWiresharkが優勝する。インストールしておくべきだ。しかし、あらゆるネットワークインシデントの最初の10分間——深く掘り下げる前にトラフィックの全体像を把握する必要があるとき——スクリプト化されたPySharkキャプチャは毎回手動フィルタリングを超える。
午前3時の応急処置として始まったスクリプトは、6ヶ月間私のサーバーで動き続けている。請求やバンド幅の上限に達するまで見逃されていたであろう異常なトラフィックパターンを3回検知した。それが成果だ：自動解析はあなたが眠っている間も休まない。
