一覧SP-06-1 · NDR：ネットワーク観測と分析

パケットとフロー情報の分析

観測点、通信量、検知条件、外部装置の措置を対応付けて評価します。

説明

NDRはネットワーク上の観測から接続や振る舞いを分析します。観測位置、暗号化、パケットの欠落によって得られる情報が変わるため、記録の項目が何を数えているかを確認します。

固定pcapをPythonで開き、アドレス、ポート、TCPデータ量を集計します。SuricataのHTTP・接続記録との対応と、NATやTLSによる見え方の違いも確認します。

観測点、通信量、検知条件、外部装置の措置を対応付けて評価します。

1取得したフレームを一覧にする

時刻、送信元、宛先、両ポート、TCPデータ長、フレーム長の順です。

from sp06 import * rows=frames取得したフレームを一覧にする() for row in rows: print(row)

6枚のフレームがあります。データが0バイトでもTCPの接続確立や確認応答として意味があります。

2接続の両方向をそろえる

端点の組を順序によらずまとめます。

pairs={tuple(sorted接続の両方向をそろえる(((s,sp),(d,dp)))) for t,s,d,sp,dp,n,frame in rows} print(sorted(pairs))

2つの方向が一つの接続にまとまります。一般にはプロトコル、開始・終了、再利用されたタプルも含めて接続を分離します。

3送信方向のデータ量を数える

端末192.0.2.7からのTCPデータ長を合計します。

print(sum送信方向のデータ量を数える(n for t,s,d,sp,dp,n,f in rows if s=='192.0.2.7')) print(sum(f for t,s,d,sp,dp,n,f in rows))

TCPデータ量と全フレーム長の合計は異なります。ヘッダーを含む転送量とアプリのデータ量を同じ尺度として扱いません。

4HTTP記録とフローを結ぶ

実解析記録にあるIDの対応を調べます。

events=RESULTS['split-alert'] http_event=next(e for e in events if e['event_type']=='http') flow_event=next(e for e in events if e['event_type']=='flow') print(http_event['flow_idHTTP記録とフローを結ぶ']==flow_event['flow_id']) print(http_event['http']['hostname'],http_event['http']['url'])

同じflow_idで接続とHTTPの情報を対応付けられます。異なる製品の任意のID同士が同じ値になるという意味ではありません。

5NAT後の観測を端末へ対応付ける

固定の変換記録に内側と外側の端点を用意します。

mapping={('203.0.113.9',61000):('192.0.2.7',53000)} print(mapping.getNAT後の観測を端末へ対応付ける(('203.0.113.9',61000))) print(mapping.get(('203.0.113.9',61001)))

61000は対応しますが61001は不明です。共有アドレスだけを同じ利用者へ割り当てず、時刻とポートを含む変換記録を使います。

6暗号化された区間の可視性を確認する

TLSの暗号文と、終端した側の平文を比較します。

from sp05 import tls_exchange暗号化された区間の可視性を確認する version,wire,plain=tls_exchange() print(version,b'Host:' in wire,b'Host:' in plain)

観測点がTLSを終端しなければHTTPヘッダーはこの暗号文から読めません。接続先や量などの情報と本文の可視性を分けて扱います。

7欠落を含む記録の結論を確認する

後半欠落pcapのHTTP記録を開きます。

print([e['http'].get欠落を含む記録の結論を確認する('url') for e in RESULTS['gap-alert'] if e['event_type']=='http'])

/lessという途中の値です。観測記録だけで完全なURLを補完して分類すると、根拠にない情報を足してしまいます。

まとめ

バイト数はどの層を数えた値か確認します。

接続IDとNAT記録で観測点間を照合します。

暗号化や欠落による未観測範囲を残します。

操作履歴と解説を振り返れます。

Python 3.15 / 固定pcap・Suricata 7.0.10解析記録

1
from sp06 import * rows=frames() for row in rows: print(row)
(1700000001, '192.0.2.7', '198.51.100.20', 53000, 80, 0, 54)
(1700000002, '198.51.100.20', '192.0.2.7', 80, 53000, 0, 54)
(1700000003, '192.0.2.7', '198.51.100.20', 53000, 80, 0, 54)
(1700000004, '192.0.2.7', '198.51.100.20', 53000, 80, 9, 63)
(1700000005, '192.0.2.7', '198.51.100.20', 53000, 80, 40, 94)
(1700000006, '198.51.100.20', '192.0.2.7', 80, 53000, 0, 54)
2
pairs={tuple(sorted(((s,sp),(d,dp)))) for t,s,d,sp,dp,n,frame in rows} print(sorted(pairs))
[(('192.0.2.7', 53000), ('198.51.100.20', 80))]
3
print(sum(n for t,s,d,sp,dp,n,f in rows if s=='192.0.2.7')) print(sum(f for t,s,d,sp,dp,n,f in rows))
49
373
4
events=RESULTS['split-alert'] http_event=next(e for e in events if e['event_type']=='http') flow_event=next(e for e in events if e['event_type']=='flow') print(http_event['flow_id']==flow_event['flow_id']) print(http_event['http']['hostname'],http_event['http']['url'])
True
demo.example.test /lesson
5
mapping={('203.0.113.9',61000):('192.0.2.7',53000)} print(mapping.get(('203.0.113.9',61000))) print(mapping.get(('203.0.113.9',61001)))
('192.0.2.7', 53000)
None
6
from sp05 import tls_exchange version,wire,plain=tls_exchange() print(version,b'Host:' in wire,b'Host:' in plain)
TLSv1.3 False True
7
print([e['http'].get('url') for e in RESULTS['gap-alert'] if e['event_type']=='http'])
['/less']

演習完了