VPCフローログをDuckDBで分析して、0.0.0.0/0のセキュリティグループを絞ってみた

こんにちは、林です。

とある案件で、0.0.0.0/0 で開いているセキュリティグループを絞り込む対応をしました。

いきなり絞って通信を止めるわけにはいかないので、VPCフローログに記録されている実際の通信から「本当に使われている範囲」を確認してから絞ります。

フローログの分析というとAthenaが定番です。
しかし今回の環境はお客様のAWSアカウントなので、分析のためだけにテーブルなどのリソースを作ったり、スキャン量次第で変わる課金を発生させたりは避けたいところです。
そこで、ログをローカルに取得してDuckDBで分析しました。

本記事では、フローログに記録された通信からセキュリティグループの許可範囲を割り出して、是正するまでの手順を紹介します。

DuckDBでのフローログ分析の基本は、こちらの記事を参考にさせていただきました。
https://dev.classmethod.jp/articles/query-vpc-flow-logs-by-duckdb/

今回の環境

社内向けBIツールが動くEC2に、こんなセキュリティグループが付いていました。

本記事は、次の2点が用意できている前提で進めます。

• VPCフローログがS3に出力されていること
• S3にアクセスできる環境（今回はAWS CloudShell）と、DuckDBを動かすローカル環境

次の流れで進めていきます。

1. 対象のセキュリティグループが付いているENIを特定する
2. フローログをS3からローカルに取得する
3. DuckDBで読み込む
4. 対象を絞ってParquet化し、ACCEPTされた通信を集計する
5. 集計結果からルール案を作り、セキュリティグループを絞る
6. 絞った後にREJECTが出ていないか確認する

手順1: セキュリティグループが付いているENIを特定する

フローログにはセキュリティグループのIDが記録されません。記録されるのはENI単位です。

まず対象のセキュリティグループがどのENIに付いているかを調べます。
あわせて、各ENIに付いているセキュリティグループも見ておきます。

aws ec2 describe-network-interfaces \
  --filters Name=group-id,Values=<セキュリティグループID> \
  --query "NetworkInterfaces[].{eni:NetworkInterfaceId,ip:PrivateIpAddress,sgs:Groups[].GroupId}"

[
    {
        "eni": "eni-0aaaa1111bbbb2222c",
        "ip": "10.0.123.13",
        "sgs": ["<セキュリティグループID>", "sg-0111111111aaaaaaa", "sg-0222222222bbbbbbb"]
    },
    {
        "eni": "eni-0dddd3333eeee4444f",
        "ip": "10.0.223.13",
        "sgs": ["<セキュリティグループID>", "sg-0333333333ccccccc"]
    }
]

出力のとおり、1つのENIには複数のセキュリティグループが付く場合があります。フローログはENI単位の記録なので、どのセキュリティグループが許可したかまでは分かりません。

そこで手がかりにするのが宛先ポート（dstport）です。今回は8080/9001番ポートを開けているのが対象のセキュリティグループだったので、このポート宛の通信を対象に関係する通信として切り出せます。この後の手順ではこのポートで絞り込みます。

手順2: フローログをローカルに取得する

S3からのログ取得はAWS CloudShellで、分析は手元のローカル環境で行いました。CloudShellはブラウザから使えるターミナルで、AWS CLIが認証済みで入っているため、認証情報をローカルに用意しなくてもそのまま aws コマンドでS3にアクセスできます。

まず、分析したい期間のログをCloudShellに取得します。

aws s3 sync "s3://<フローログのバケット>/AWSLogs/123456789012/vpcflowlogs/ap-northeast-1/" ./flowlogs/ \
  --exclude "*" --include "*2026-06*" --include "*2026-07*"

フローログは日付ごとにディレクトリが分かれて出力されるので、--include で対象の期間だけに絞れます。

s3://<バケット>/AWSLogs/123456789012/vpcflowlogs/ap-northeast-1/
├── 2026/06/10/
│   └── 123456789012_vpcflowlogs_ap-northeast-1_fl-0abc..._20260610T0000Z_xxxx.log.gz
├── 2026/06/11/
│   └── ...
└── 2026/07/10/
    └── ...

今回対象とした約1ヶ月分は、gzip圧縮された状態で合計100MB程度でした。

取得したログは、CloudShellの「Actions」→「Download file」からローカルにダウンロードします。ファイル数が多いので、tarでまとめてから1ファイルとしてダウンロードすると楽になります。

tar czf flowlogs.tar.gz flowlogs/

手順3: DuckDBで読み込む

ここからはローカルでの作業です。ダウンロードしたログを展開しておきます。

tar xzf flowlogs.tar.gz

DuckDBが入っていなければ、公式のインストールスクリプトで入れます。
（DuckDB Installation）

curl https://install.duckdb.org | sh

インストール後、duckdb と打つと対話用のプロンプトが起動し、そこにSQLを直接入力して実行できます。SQLをファイルにまとめておけば、duckdb -c ".read analysis.sql" のようにまとめて実行することもできます。

ここからは、ログファイルにSQLで問い合わせられる状態を作っていきます。
やることは2つです。

1. ログの各列に名前を付ける（action や srcaddr などSQLで扱いやすくする）
2. 分析に使う分だけ抜き出して、扱いやすい形式で保存する

ログの各列に名前を付ける

フローログの各行はスペース区切りのテキストで、列名は付いていません。
まずどの位置に何のフィールドが入っているかを確認します。

デフォルト形式ならフィールドの並びは固定ですが、カスタム形式だと環境ごとに違います。describe-flow-logs で確認しましょう。

aws ec2 describe-flow-logs --query "FlowLogs[].LogFormat"

今回の環境は、こんな29フィールドのカスタム形式でした。
（フローログ作成時に「すべての属性」を選ぶとこの形式になります。）

[
  "${account-id} ${action} ${az-id} ${bytes} ${dstaddr} ${dstport} ${end} ${flow-direction} ${instance-id} ${interface-id} ${log-status} ${packets} ${pkt-dst-aws-service} ${pkt-dstaddr} ${pkt-src-aws-service} ${pkt-srcaddr} ${protocol} ${region} ${srcaddr} ${srcport} ${start} ${sublocation-id} ${sublocation-type} ${subnet-id} ${tcp-flags} ${traffic-path} ${type} ${version} ${vpc-id}"
]

DuckDBは列名のないファイルを読むと、各列を先頭から column00, column01, ... と自動で名付けます。上のフィールドの並び順がこの番号に対応するので、必要な列だけ分かりやすい名前を付け直した「ビュー」を作っておきます。
ビューを作っておくと、以降のクエリは長い読み込み指定を書かずに flow_logs という名前で参照できます。

CREATE OR REPLACE VIEW flow_logs AS
SELECT
  column01 AS action,        -- ACCEPT / REJECT
  CAST(column03 AS BIGINT) AS bytes,
  column04 AS dstaddr,       -- 宛先IP
  TRY_CAST(column05 AS INT) AS dstport,   -- 宛先ポート
  column07 AS flow_direction,
  column09 AS interface_id,  -- ENI
  column10 AS log_status,
  TRY_CAST(column16 AS INT) AS protocol,
  column18 AS srcaddr,       -- 送信元IP
  to_timestamp(CAST(column20 AS BIGINT)) AS start_time
  -- 実際のLogFormatに合わせて調整してください
FROM read_csv('flowlogs/**/*.log.gz', delim=' ', header=false, all_varchar=true);

SQLの上から順に、使っている変換処理と読み込みオプションの意味を挙げます。

• CAST(...)：文字列を数値型に変換するSQLの操作で、集計や比較に使う列に用いる
• TRY_CAST(...)：CAST の安全版で、-（記録なし）のように変換できない値はエラーにせず空（NULL）として扱う（ポートなどの列で必要）
• to_timestamp(...)：UNIX秒で記録された時刻を、読みやすい日時に変換する
• FROM read_csv(...)：flowlogs/ 配下の .log.gz をまとめて読み込む（header=false はヘッダー行なし、all_varchar=true は全列を文字列として読み込む指定）

<version> <account-id> <interface-id> <srcaddr> <dstaddr> <srcport> <dstport> <protocol> <packets> <bytes> <start> <end> <action> <log-status>

CREATE OR REPLACE VIEW flow_logs AS
SELECT
  column02 AS interface_id,  -- ENI
  column03 AS srcaddr,       -- 送信元IP
  column04 AS dstaddr,       -- 宛先IP
  TRY_CAST(column06 AS INT) AS dstport,   -- 宛先ポート
  TRY_CAST(column07 AS INT) AS protocol,
  CAST(column09 AS BIGINT) AS bytes,
  to_timestamp(CAST(column10 AS BIGINT)) AS start_time,
  column12 AS action,        -- ACCEPT / REJECT
  column13 AS log_status
FROM read_csv('flowlogs/**/*.log.gz', delim=' ', header=false, all_varchar=true);

分析対象だけ抜き出してParquetに保存する

ログ全体を毎回読むと時間がかかります。
そこで、今回分析したい範囲（対象ENIの、8080/9001番ポート宛の受信通信）だけを抜き出し、target_flows.parquet というファイルに保存します。
Parquetは列単位で高速に読める形式で、以降のクエリはこのファイルに対して実行します。

COPY (
  SELECT * FROM flow_logs
  WHERE flow_direction = 'ingress'   -- 受信方向
    AND log_status = 'OK'            -- 記録あり
    AND interface_id IN ('eni-0aaaa1111bbbb2222c', 'eni-0dddd3333eeee4444f')
    AND protocol = 6                 -- TCP
    AND dstport IN (8080, 9001)
) TO 'target_flows.parquet' (FORMAT PARQUET);

WHERE で指定している条件は、それぞれ次の意味です。

• flow_direction = 'ingress'：外から入ってくる受信方向の通信だけに絞る
• log_status = 'OK'：NODATA / SKIPDATA（通信そのものの記録ではない行）を除外する
• interface_id IN (...)：手順1で特定した対象ENIに限定する
• protocol = 6：TCPに限定する（6はTCPを表すプロトコル番号）
• dstport IN (8080, 9001)：対象のセキュリティグループが開けているポート宛だけに絞る

なお、フローログの1行は個々のパケットではなく、一定時間内の似た通信をまとめた「フロー」という単位です。ここでいう件数は、このフローの数を指します。

この絞り込みの結果、今回は212,158フローが対象になりました。
約100MBのログの読み込みから target_flows.parquet の出力まで、約1.5秒です。
以降の集計はこの絞り込み済みファイルに対して行うので、さらに速くなります。

手順4: ACCEPTされた通信を集計する

まず、許可された通信（action = 'ACCEPT'）を、送信元IPと宛先ポートの組み合わせごとに集計します。
フロー件数の多い順に、上位10件を見てみます。

SELECT srcaddr, dstport, count(*) AS flows, sum(bytes) AS total_bytes
FROM 'target_flows.parquet'
WHERE action = 'ACCEPT'
GROUP BY srcaddr, dstport
ORDER BY flows DESC
LIMIT 10;

action = 'ACCEPT' で許可された通信だけに絞り、送信元IPと宛先ポートの組み合わせごとに、フロー件数（count(*)）と通信量（sum(bytes)）を集計しています。
件数の多い順に上位10件を表示します。

上位はすべて 10.100. で始まる社内ネットワークのIPでした。ただ、送信元IPは種類が多く、上位10件を見るだけでは全体像はつかめません。
そこで、ENIとポートごとに「送信元IPが何種類あるか」を数えます。重複を除いた件数は count(DISTINCT srcaddr) で求められます。

SELECT f.interface_id, f.dstport, count(*) AS flows,
       count(DISTINCT f.srcaddr) AS unique_sources
FROM 'target_flows.parquet' f
WHERE f.action = 'ACCEPT'
GROUP BY ALL
ORDER BY flows DESC;

本番側（eni-0aaaa1111bbbb2222c）の8080番ポートには、1ヶ月間で約2,500のユニークな送信元から20万件を超えるフローが記録されていました。送信元をIP単位で1つずつ確認するのは現実的ではないので、次の手順ではネットワーク単位にまとめて整理します。

手順5: 集計結果からルール案を作る

送信元IPを1つずつ見るのではなく、社内ネットワークの区分（社内NW・VPN・AWS環境）ごとにまとめて集計します。

SELECT dstport,
       CASE
         WHEN srcaddr LIKE '10.100.%' THEN '10.100.0.0/16 (社内NW)'
         WHEN srcaddr LIKE '10.200.%' THEN '10.200.0.0/16 (VPN A)'
         WHEN srcaddr LIKE '10.210.%' THEN '10.210.0.0/16 (VPN B)'
         WHEN srcaddr LIKE '10.0.%'   THEN '10.0.0.0/16 (AWS環境)'
         ELSE 'その他'
       END AS src_range,
       count(*) AS flows, count(DISTINCT srcaddr) AS unique_ips
FROM 'target_flows.parquet'
WHERE action = 'ACCEPT'
GROUP BY ALL ORDER BY dstport, flows DESC;

CASE 式で、送信元IPをネットワークごとに分類しています。
（LIKE '10.100.%' は「10.100. で始まるIP」の意味で、どのレンジにも当てはまらないものは ELSE で「その他」にまとめます。）

ここで注目すべきは、「その他」が0件だったことです。
0.0.0.0/0 で開けてはいたものの、1ヶ月間にACCEPTされた通信は、すべて社内・VPN・AWS内のIPレンジからのもので、インターネットからのアクセスはありませんでした。

つまり、この4つのレンジだけを許可すれば、実際の通信を止めずに 0.0.0.0/0 をなくせます。
ルール案は次のとおりです。

絞る前の確認ポイントです。

• ログの観測期間が業務サイクルの1周分以上あるか
  （月次バッチのような低頻度の通信は短い期間だと現れないため、今回は1ヶ月分を確認）
• 送信元がごく少数のIPに限られる場合は、レンジ（/16 など）ではなく /32（そのIPだけ）まで絞る選択肢もある
  （今回の9001番ポートのAWS環境内は1IPだけだった）
• ELBやNAT越しの通信は、送信元IPの見え方が変わる点に注意

あとは、この案に沿って 0.0.0.0/0 のルールを削除し、必要なレンジのルールを追加すれば是正完了です。是正後は describe-security-group-rules で結果を確認しておきます。

aws ec2 describe-security-group-rules \
  --filters Name=group-id,Values=<セキュリティグループID> \
  --query "sort_by(SecurityGroupRules[?IsEgress==\`false\`],&FromPort)[].{proto:IpProtocol,port:FromPort,cidr:CidrIpv4,desc:Description}"

[
    {"proto": "tcp", "port": 8080, "cidr": "10.100.0.0/16", "desc": "Internal NW"},
    {"proto": "tcp", "port": 8080, "cidr": "10.200.0.0/16", "desc": "VPN A"},
    {"proto": "tcp", "port": 8080, "cidr": "10.210.0.0/16", "desc": "VPN B"},
    {"proto": "tcp", "port": 8080, "cidr": "10.0.0.0/16", "desc": "AWS"},
    {"proto": "tcp", "port": 9001, "cidr": "10.100.0.0/16", "desc": "Internal NW"},
    {"proto": "tcp", "port": 9001, "cidr": "10.0.0.0/16", "desc": "AWS"}
]

0.0.0.0/0 が消え、通信のあったレンジだけが残りました。

是正後の確認について

セキュリティグループを絞ったあとは、しばらく運用してから、今度は拒否された通信（REJECT）を確認しておくと安心です。正当な通信を誤って塞いでいた場合、ここにREJECTとして現れます。
手順2〜3と同じ要領で是正後の期間のログを取得し、ビュー（flow_logs）を作り直してから、次のクエリを実行します。

SELECT srcaddr, dstport, count(*) AS rejects
FROM flow_logs
WHERE action = 'REJECT' AND flow_direction = 'ingress'
GROUP BY srcaddr, dstport
ORDER BY rejects DESC;

action = 'REJECT' で拒否された通信だけに絞り、送信元IPと宛先ポートごとに件数を数えています。想定外のREJECTが出たら、その送信元が正当なものかを確認し、必要ならルールに追加します。是正後の確認も、これまでと同じDuckDBでの集計だけで済みます。

Athenaとの比較

今回の100MB程度なら、Athenaでも1クエリ0.1円未満と、料金の差はほぼありませんでした。それでもDuckDBにしたことで、お客様アカウントに分析用のリソースを作らず、スキャン課金も気にせずに済みました。ログを取得したあとは、何度クエリを打ち直してもお客様環境には影響しません。

まとめ

• 0.0.0.0/0 を絞るときは、フローログでACCEPTされた通信を見れば、必要な許可範囲がデータで分かる
• 棚卸しのような一時的な分析なら、DuckDBで十分（約100MBを約1.5秒、費用もほぼかからない）
• お客様アカウントにリソースを作らず、課金も気にせず分析できる

分析時のハマりどころもまとめておきます。

• dstport = 0 の行はICMP（ポートの概念がないため0になる）なので、protocol 列と合わせて確認する
• ENIとセキュリティグループは1対1ではないため、どの通信がどのルールで許可されたかを対応づけるには、ポートとルールの突き合わせが必要
• カスタム形式のログは、列の対応付けが必要（手順3参照）
• フローログのREJECTは、セキュリティグループとNACLのどちらで拒否されたか区別できない

さいごに

最近は、こうした分析から是正案の作成まで、AIにまとめて任せられるようになってきました。
今回は、仕組みを自分で理解しておきたかったので、あえて手を動かしてやってみました。

この記事がどなたかの参考になれば幸いです。最後までご覧いただきありがとうございました！

参考

https://dev.classmethod.jp/articles/query-vpc-flow-logs-by-duckdb/

https://dev.classmethod.jp/articles/vpc-flow-log-default-field/

https://docs.aws.amazon.com/vpc/latest/userguide/flow-logs.html