通信量と通信間隔の分析
固定のTCP観測列を方向別に集計し、シーケンス番号が同じ再送を確認します。周期的な接続も業務のヘルスチェックと比較し、量や周期だけで用途を断定しない方法を学びます。
説明
通信量は、IPパケット全体の量、TCPのペイロード、重複を除いたデータ量で値が変わります。再送を含むパケットの合計を、そのまま送ったファイルのサイズとはみなせません。
固定のTCP観測列を方向別に集計し、シーケンス番号が同じ再送を確認します。周期的な接続も業務のヘルスチェックと比較し、量や周期だけで用途を断定しない方法を学びます。
1pcapのヘッダーと先頭フレームを開く
このpcapはTCP接続中の4パケットを収めた固定試料です。先頭24バイトはpcap全体のヘッダーで、その後に各パケットの記録ヘッダーとフレームが続きます。
from oppacket import *
raw=read_capture()
print("pcap:",raw[:24].hexpcapのヘッダーと先頭フレームを開く(" "))
first=next(frames())[1]
print("Ethernet:",first[:14].hex(" "))
print("IPv4:",first[14:34].hex(" "))
print("TCP:",first[34:54].hex(" "))
print("payload:",first[54:].hex(" "),first[54:])EtherType 08 00はIPv4、IPv4先頭45はバージョン4・ヘッダー20バイトです。TCPのシーケンス番号は00 00 00 64、ペイロード41 42 43 44はASCIIのABCDです。
2ヘッダー長を使ってデータを抽出する
decodeはIPv4のIHLとTCPのData Offsetから境界を計算します。各行は番号、秒、接続ID、方向、シーケンス番号、データ長、IP長です。
rows=decodeヘッダー長を使ってデータを抽出する()
for row in rows: print(row)送信1回目と2回目はseq=100で4バイト、3回目はseq=104で2バイトです。以後の集計表は、このpcapから得た列と一致する固定資料を使います。
3パケットごとの値を見る
payloadはTCPデータの長さ、ip_bytesはヘッダー込みです。試料ではIPとTCPのヘッダーがそれぞれ20バイトです。
from oplab import *
db=connect(5)
show(db, 'SELECTパケットごとの値を見る * FROM packets ORDER BY id')1番と2番は同じシーケンス番号100で4バイトを送っています。この試料では同じ内容の再送です。
4方向別に観測量を計算する
IPの長さとペイロード長を別に足します。
show(db, 'SELECT方向別に観測量を計算する direction,SUM(ip_bytes) AS ip_bytes,SUM(payload) AS payload_bytes FROM packets GROUP BY direction ORDER BY direction')送信側は130バイトのIPパケット、ペイロードの延べ長は10バイトです。どの層の量かを添えて説明します。
5完全に重なる区間を除く
この試料は区間の重なりが完全一致する例です。一般のTCP再構成には部分重複やシーケンス番号の周回も考慮します。
show(db, 'SELECT完全に重なる区間を除く SUM(payload) AS distinct_payload FROM (SELECT DISTINCT seq,payload FROM packets WHERE direction="out")')一意な送信データは6バイトです。この単純なDISTINCTを任意のpcapに適用して正しい再構成になるわけではありません。
6接続間隔を計算する
同じ端末の前回時刻をLAGで取り出し、差を計算します。
show(db, 'SELECT接続間隔を計算する host,id,t,t-LAG(t) OVER(PARTITION BY host ORDER BY t) AS gap FROM periodic ORDER BY host,t')両端末とも60秒間隔です。一定間隔という特徴は、悪意のある通信と正常な監視処理のどちらにも現れます。
7業務処理との対応を確認する
jobは運用台帳との照合結果です。NULLはこの資料で対応が取れていないことを示します。
show(db, 'SELECT業務処理との対応を確認する host,job,COUNT(*) AS connections,SUM(bytes) AS bytes FROM periodic GROUP BY host,job ORDER BY host')ws8はhealthcheckに対応します。ws7は用途未確認ですが、それだけで攻撃用通信と確定するわけではありません。
8用途が未確認の周期を残す
追加調査する端末の観測量をまとめます。
show(db, 'SELECT用途が未確認の周期を残す host,MIN(t) AS first,MAX(t) AS last,COUNT(*) AS count FROM periodic WHERE job IS NULL GROUP BY host')ws7の3接続を、実行プロセス、宛先、通常時の記録へ照合する調査が残ります。数値は候補を絞る材料として扱います。
まとめ
ws7の3接続を、実行プロセス、宛先、通常時の記録へ照合する調査が残ります。数値は候補を絞る材料として扱います。
表示された証拠IDと検索条件を残すと、調査結果を元の記録までたどれます。
操作履歴と解説を振り返れます。
Python 3.15 / pcap・Ethernet/IPv4/TCP・SQLite
from oppacket import *
raw=read_capture()
print("pcap:",raw[:24].hex(" "))
first=next(frames())[1]
print("Ethernet:",first[:14].hex(" "))
print("IPv4:",first[14:34].hex(" "))
print("TCP:",first[34:54].hex(" "))
print("payload:",first[54:].hex(" "),first[54:])pcap: d4 c3 b2 a1 02 00 04 00 00 00 00 00 00 00 00 00 ff ff 00 00 01 00 00 00 Ethernet: 02 00 00 00 00 08 02 00 00 00 00 07 08 00 IPv4: 45 00 00 2c 00 01 40 00 40 06 3c bc c0 00 02 07 cb 00 71 07 TCP: c7 38 01 bb 00 00 00 64 00 00 00 c8 50 18 10 00 53 13 00 00 payload: 41 42 43 44 b'ABCD'
rows=decode()
for row in rows: print(row)(1, 0.0, 'c7', 'out', 100, 4, 44) (2, 0.2, 'c7', 'out', 100, 4, 44) (3, 0.3, 'c7', 'out', 104, 2, 42) (4, 0.4, 'c7', 'in', 200, 3, 43)
from oplab import *
db=connect(5)
show(db, 'SELECT * FROM packets ORDER BY id')id | t | flow | direction | seq | payload | ip_bytes 1 | 0.0 | c7 | out | 100 | 4 | 44 2 | 0.2 | c7 | out | 100 | 4 | 44 3 | 0.3 | c7 | out | 104 | 2 | 42 4 | 0.4 | c7 | in | 200 | 3 | 43
show(db, 'SELECT direction,SUM(ip_bytes) AS ip_bytes,SUM(payload) AS payload_bytes FROM packets GROUP BY direction ORDER BY direction')direction | ip_bytes | payload_bytes in | 43 | 3 out | 130 | 10
show(db, 'SELECT SUM(payload) AS distinct_payload FROM (SELECT DISTINCT seq,payload FROM packets WHERE direction="out")')distinct_payload 6
show(db, 'SELECT host,id,t,t-LAG(t) OVER(PARTITION BY host ORDER BY t) AS gap FROM periodic ORDER BY host,t')host | id | t | gap ws7 | c1 | 0 | NULL ws7 | c2 | 60 | 60 ws7 | c3 | 120 | 60 ws8 | c4 | 0 | NULL ws8 | c5 | 60 | 60 ws8 | c6 | 120 | 60
show(db, 'SELECT host,job,COUNT(*) AS connections,SUM(bytes) AS bytes FROM periodic GROUP BY host,job ORDER BY host')host | job | connections | bytes ws7 | NULL | 3 | 180 ws8 | healthcheck | 3 | 180
show(db, 'SELECT host,MIN(t) AS first,MAX(t) AS last,COUNT(*) AS count FROM periodic WHERE job IS NULL GROUP BY host')host | first | last | count ws7 | 0 | 120 | 3
演習完了