はじめに：夜間障害の現場で、私たちは何を見ているのか？
深夜2時。PagerDutyのけたたましいアラート音で叩き起こされ、ダッシュボードを開くと、あるマイクロサービスのレイテンシが異常に跳ね上がっている。APIサーバーのCPU使用率は正常、DBのコネクション数も枯渇していない。では、何が起きているのか？
「おい、ネットワークの帯域がどこかで飽和してるぞ。誰が何を踏み抜いてる？」
こういう修羅場で、私たちシニアSREが真っ先に手を伸ばすのが VPCフローログ（VPC Flow Logs） です。CloudWatch LogsやS3のバケツに吐き出される、あの無機質なCSVの羅列。しかし、その一見ただの数字の奔流には、パケットがAWSの仮想ルーターをどう駆け抜け、どのENI（Elastic Network Interface）でドロップされたのかという「真実のログ」が刻まれています。
今回は、VPCフローログのレコードフォーマット、特に srcaddr、dstaddr、packets、bytes という最も重要で、かつ誤解されやすい4つのフィールドに焦点を当て、不正アクセスの炙り出しから帯域泥棒の特定まで、現場で即座に使える実務的解析手法を叩き込みます。
—
1. VPCフローログの基本構造と「バージョン2」の罠
まずは、AWSのVPCフローログがどのようなフォーマットで出力されているのかを確認しましょう。デフォルト（バージョン2）で出力されるレコードは、スペース区切りのプレーンテキストであり、以下のような並び順を持っています。
2 123456789012 eni-0123456789abcdef0 192.0.2.1 198.51.100.5 443 52342 6 12 1500 1680000000 1680000060 ACCEPT OKそれぞれのフィールドが何を意味しているのか、左から順に紐解いていきましょう。
1. version: フローログのフォーマットバージョン（通常は 2）
2. account-id: AWSアカウントID
3. interface-id: ログ対象のENI ID
4. srcaddr: パケットの送信元IPアドレス
5. dstaddr: パケットの宛先IPアドレス
6. srcport: 送信元ポート番号
7. dstport: 宛先ポート番号
8. protocol: IANAプロトコル番号（6 は TCP、17 は UDP）
9. packets: この集約期間中に転送されたパケット数
10. bytes: この集約期間中に転送されたバイト数の合計
11. start: フローの開始エポック秒
12. end: フローの終了エポック秒
13. action: アクセス制御の結果（ACCEPT または REJECT）
14. log-status: ログの状態（OK、NODATA、SKIPDATA）
ここで、多くのエンジニアがやりがちな最初の勘違いがあります。それは、srcaddr や dstaddr が「常に外から来たクライアントのIPと、WebサーバーのプライベートIPだ」と思い込むことです。
パケットの「向き（Direction）」を意識せよ
VPCフローログは、ENI単位でトラフィックを記録します。つまり、ログに記録される srcaddr と dstaddr は、そのENIから見たときの「そのパケットの送信元・宛先」 であり、通信の全体像（クライアント視点なのかサーバー視点なのか）は、ENIがアタッチされているリソース（EC2、NATゲートウェイ、ELBなど）によって意味合いが変わります。
例えば、パブリックサブネットにあるALB（Application Load Balancer）のENIを見る場合：
- インバウンド通信（ユーザーからのリクエスト）：
srcaddrはエンドユーザーのグローバルIP、dstaddrはALBのプライベートIPになります。 - アウトバウンド通信（ALBからバックエンドEC2への転送）：
srcaddrはALBのプライベートIP、dstaddrはバックエンドEC2のプライベートIPになります。
この「視点（Point of View）」を誤ると、不正アクセス元を特定するつもりが、自社のNATゲートウェイのIPをブロックしてしまうという大惨事を引き起こします。
—
2. コアフィールドの深掘り：srcaddr, dstaddr, packets, bytes
それでは、今回のテーマの核心である4つのフィールドについて、パケットの挙動と結びつけて深く掘り下げてみましょう。
srcaddr と dstaddr：IPアドレスの「非対称性」に注意する
TCP/IPの通信において、リクエストとレスポンスは往復します。しかし、VPCフローログは単方向のフローごとに集約されて記録されます（厳密には、セキュリティグループのステートフルな挙動や、AWS内部のネットワーク処理に依存しますが、基本は双方向が別々のレコード、あるいは一定期間の集約値として現れます）。
特に注意すべきは、AWSのマネージドサービス（NAT GatewayやENIなど）を挟む場合です。例えば、プライベートサブネットのEC2から外部APIへHTTPS通信を行った場合、フローログの srcaddr はEC2のプライベートIPですが、インターネット側のルーターから見ると、それはNAT GatewayのElastic IP（EIP）に変換（SNAT）されています。
したがって、セキュリティインシデントの調査で「どの内部サーバーが外部の怪しいIPと通信しているか」を追うときは、NAT Gatewayのフローログを逆引きできるように、ENIの紐付けを正しく把握しておく必要があります。
packets と bytes：帯域泥棒とDDoSの兆候を掴む
次に packets と bytes です。この2つの値の関係性（平均パケットサイズ）を見ることで、その通信がどのような性質のものかを一発で見抜くことができます。
bytes/packetsの比率が非常に小さい場合（小パケットの洪水）:
1パケットあたりのバイト数が数十バイトしかないのに、packets の数が異常に多い場合、これは典型的な DDoS攻撃（SYNフラッドやUDP反射攻撃） や、細切れのリクエストを大量に送りつけるアプリケーションのバグ（あるいはスクレイピング・ボットの猛攻）を疑うべきです。
bytesが巨大で、packetsも多い場合:
動画配信、大規模なデータベースのレプリケーション、あるいはS3からの大容量オブジェクトダウンロードなどが該当します。もし意図しない時間帯にこのメトリクスが跳ね上がっているなら、不正なデータ持ち出し（データエクスフィルトレーション）や、バッチ処理の暴走を疑う必要があります。
—
3. 実践：PythonでVPCフローログを解析し、異常なトラフィックを炙り出す
AWSのコンソールやAthenaでクエリを書くのも良いですが、S3に溜まった膨大なGzip圧縮されたフローログをローカルやLambdaで素早く解析したい場面は多々あります。
ここでは、Pythonを使用して、S3からダウンロードしたVPCフローログをパースし、「特定の宛先に対して異常なバイト数を送信している上位IP（帯域泥棒）」 と 「REJECTされた回数が異常に多い不審なアクセス元（不正アクセス検知）」 を検出するスクリプトを紹介します。
実務でそのまま使えるよう、例外処理やデータ構造の整理を意識したコードに仕上げています。
import gzip
import io
from collections import defaultdict
import urllib.request
def parse_vpc_flow_log(log_data_bytes):
"""
VPCフローログのバイナリデータ（Gzip展開前）を受け取り、
パースして集計用データを返す関数
"""
# 統計情報の格納用辞書
traffic_by_ip = defaultdict(lambda: {"packets": 0, "bytes": 0})
rejected_attempts = defaultdict(int)
# Gzipの解凍
try:
with gzip.GzipFile(fileobj=io.BytesIO(log_data_bytes)) as f:
for line in f:
decoded_line = line.decode('utf-8').strip()
if not decoded_line:
continue
parts = decoded_line.split()
# バージョン2の標準フォーマット長（14カラム以上）を検証
if len(parts) < 14:
continue
# スキップデータやNODATAは除外
if parts[13] in ("NODATA", "SKIPDATA"):
continue
version = parts[0]
# バージョンに応じたパース（今回はV2を前提）
if version == "2":
src_addr = parts[3]
dst_addr = parts[4]
try:
packets = int(parts[8])
bytes_transferred = int(parts[9])
except ValueError:
continue # 数値変換エラーのレコードはスキップ
action = parts[12]
# 1. 帯域消費量の集計（送信元IPごとのトータルバイト数・パケット数）
traffic_by_ip[src_addr]["packets"] += packets
traffic_by_ip[src_addr]["bytes"] += bytes_transferred
# 2. 不正アクセス（REJECT）の集計
if action == "REJECT":
rejected_attempts[src_addr] += 1
except Exception as e:
print(f"ログのパース中にエラーが発生しました: {e}")
return traffic_by_ip, rejected_attempts
def analyze_network_bottlenecks(traffic_by_ip, rejected_attempts, byte_threshold=1000000000, reject_threshold=100):
"""
集計結果をしきい値と比較し、ボトルネックやセキュリティインシデントの候補を出力する
"""
print("=== 【帯域消費アラート】大容量トラフィックを送信している上位IP ===")
# バイト数の降順でソートして上位を表示
sorted_traffic = sorted(traffic_by_ip.items(), key=lambda x: x[1]["bytes"], reverse=True)
for ip, stats in sorted_traffic[:5]:
total_gb = stats["bytes"] / (1024 * 1024 * 1024)
print(f"IP: {ip} | 総バイト数: {total_gb:.2f} GB | 総パケット数: {stats['packets']}")
if stats["bytes"] > byte_threshold:
print(f" -> 【警告】しきい値 ({byte_threshold} bytes) を超える大容量通信を検出しました。")
print("\n=== 【セキュリティアラート】拒否（REJECT）された試行回数が多いIP ===")
sorted_rejects = sorted(rejected_attempts.items(), key=lambda x: x[1], reverse=True)
for ip, count in sorted_rejects[:5]:
print(f"IP: {ip} | REJECT回数: {count}")
if count > reject_threshold:
print(f" -> 【警告】不正アクセス（ポートスキャン等）の可能性が高いIPです。")
# --- 実行モック（実運用ではS3クライアントからバイナリを取得して渡す） ---
if __name__ == "__main__":
# テスト用のダミーログデータ（実際にはS3から取得したGzipデータ）
# 形式: version account eni src dst sport dstport proto packets bytes start end action status
dummy_log_content = (
"2 123456789012 eni-123 192.0.2.50 10.0.1.10 54321 443 6 50000 10737418240 1680000000 1680000060 ACCEPT OK\n"
"2 123456789012 eni-123 203.0.113.99 10.0.1.15 12345 80 6 15000 500000 1680000000 1680000060 REJECT OK\n"
)
# バイト列に変換してGzip圧縮
out = io.BytesIO()
with gzip.GzipFile(fileobj=out, mode='w') as f:
f.write(dummy_log_content.encode('utf-8'))
gzipped_bytes = out.getvalue()
# 解析実行
traffic, rejects = parse_vpc_flow_log(gzipped_bytes)
analyze_network_bottlenecks(traffic, rejects)このスクリプトをAWS Lambdaに組み込み、S3へのログ配置イベント（s3:ObjectCreated:*）をトリガーとして実行するように構成しておけば、深夜であっても「誰がネットワークを圧迫しているか」「どこから執拗なスキャンが来ているか」をSlackに自動通知させることができます。
—
4. 現場で役立つ実践的Tips：Athenaを活用した効率的なクエリ戦略
S3に数テラバイト規模のVPCフローログが溜まってくると、Pythonでの全件走査では時間がかかりすぎます。そんなときは Amazon Athena を使って、Parquet形式やパーティション分割されたS3上のログに対してSQLを叩くのが正解です。
以下は、実務でよく使う「過去1時間で最もトラフィックを消費している送信元IPトップ10」を炙り出すAthenaクエリの模範例です。
SELECT
srcaddr,
SUM(cast(bytes as bigint)) AS total_bytes,
SUM(cast(packets as bigint)) AS total_packets,
COUNT(*) AS flow_count
FROM
vpc_flow_logs
-- パーティション（年/月/日/時）でスキャン範囲を絞り込み、コストを削減する
WHERE
year = '2023'
AND month = '10'
AND day = '27'
AND hour >= '10'
GROUP BY
srcaddr
ORDER BY
total_bytes DESC
LIMIT 10;💡 SREの現場Tips
Athenaでフローログをクエリする際、WHERE 句にパーティション（時間や日付）を指定し忘れると、S3上の全データをスキャンされてしまい、AWS利用料の請求書を見た瞬間に冷や汗をかくことになります。AWS Glueを使ってパーティション射影（Partition Projection）を必ず設定し、コスト爆発を防ぐのがプロのインフラアーキテクトの仕事です。
—
おわりに：ログを制する者は、ネットワーク障害を制する
クラウドのネットワークは、一見すると「ブラックボックス」のようによく分からない挙動をするように思われがちです。ルーターの設定も、物理的な配線も見えない。私たちが触れるのは、APIと、画面の向こうに流れるメトリクスとログだけです。
しかし、今回解説した srcaddr、dstaddr、packets、bytes というVPCフローログの基本構造をしっかりと体に叩き込んでおけば、パケットがどこから来て、どこへ消えたのか、そのストーリーを正確に描き出すことができます。
障害は突然やってきます。その時に「ログの読み方がわからない」と立ち尽くさないよう、平時のうちに自分のVPCのフローログをAthenaやスクリプトで眺め、「うちの環境の通常のトラフィックの形」を肌感覚で掴んでおいてください。その地道な積み重ねこそが、深夜の障害対応であなたを救う最大の武器となります。
コメント