【AWSのログ管理】VPC Flow LogsをAthenaを使って調査してみる

2023年5月16日 16:28

VPC Flow Logsとは

VPC Flow Logsとは、Amazon Virtual Private Cloud (VPC) のネットワークトラフィックに関する情報をキャプチャしてログとして記録するAWSのサービスです。

VPC Flow Logsを有効にすることで、インターネットゲートウェイやEC2インスタンス、Elastic Load Balancerなど、VPC内のリソースで生成されるトラフィックに関する情報を収集できます。これには、送信元IP・宛先IP、送信元ポート番号・宛先ポート番号、トラフィックのプロトコル、パケットやバイト数などが含まれます。
VPC Flow Logsは以下のような目的で利用することができます。

トラブルシューティングや問題解決時のネットワークトラフィックの分析
セキュリティグループの設定の見直しや不正なトラフィックの検出
監視、ログの保存、コンプライアンスの維持

フローログはCloudwatch LogsやS3に保存することができます。
S3に保存すると、Athenaなどの分析ツールを使って分析することができるようになります。

今回は、VPC Flow LogsをAthenaを使って分析してみたいと思います。
わかりやすく図にすると以下のような構成になります。

それでは設定を行っていきます。

ログ保存先の設定(S3バケット)

S3バケットの作成
まずは、フローログを保存するためのS3バケットを作成します。
※フローログはCloudwatchLogsに保存することもできますが、今回はAthenaでクエリをかけたいのでS3保存とします。

S3バケット作成の手順は割愛しますが、
S3バケット作成後、ARNをコピーしておくとよいでしょう。

VPC Flow Logsの設定

1.AWS コンソールにログインし、VPCを選択します。

2.VPC Flow Logsを有効化するVPCを選択します。

3.アクションプルダウンから「フローログを作成」をクリック

4.フローログの設定画面の入力

「名前」に任意の名前を入力してください。
「フィルタ」はすべてを選択
「最大集約間隔」はデフォルトのまま10分間を選択
「送信先」はAmazon S3バケットに送信を選択します。
　※Athenaでクエリをかけたいので今回はS3を選択
「S3 バケット ARN」欄に先程作成したS3バケットのARNを入力します。
　※S3のARNを事前にコピーしていた場合は、ペーストします。

「ログレコード形式」はデフォルトのままAWSのデフォルト形式を選択
「ログファイル形式」はデフォルトのまま
「Hive 互換 S3 プレフィックス」デフォルトのまま
「時間別にログをパーティション分割」デフォルトのまま
　※大量にログがあると予想される場合、1時間ごとに分割を検討しましょう

設定項目の入力は以上です。
最下段の「フローログを作成」をクリックすると設定は完了です。
※しばらく待つとログが指定したS3バケットに出力し始めます。

ログの形式について

フローログは以下のような形で出力されます。

2 123456789012 eni-01234567 10.0.0.1 172.16.0.1 41277 50219 6 1 40 1683806916 1683806926 REJECT OK

左から順番に
・VPC フローログバージョン
・AWS アカウント ID
・ENI ID
・送信元アドレス
・送信先アドレス
・送信元ポート
・送信先ポート
・プロトコル ※6はTCP
・パケットの数
・バイト数
・最初のパケットが受信された時間((UNIX 秒表記)
・最後のパケットが受信された時間((UNIX 秒表記)
・トラフィックアクション(許可がACCEPT、拒否がREJECT)
・ログのステータス(記録ができたか)

Athenaの設定

フローログの調査ができるように、Athenaを設定していきます。

1.AWS コンソールにログインし、Athenaを選択します。

2.「クエリエディタを起動」をクリック。

3.クエリ結果の保存設定
初めてAthenaを利用する場合、画像にあるような注意事項が表示され、クエリエディタからクエリを実行することができません。
Athenaのクエリ結果を保存するためのS3バケット設定が必要になります。

予めS3バケットの用意がある場合はそのまま「設定を編集」をクリックして設定にすすみます。
S3バケットの用意が無い場合は、S3バケットを新たに作成してから画像にある「設定を編集」にすすみます。

「設定を編集」をクリック後、いくつか入力項目があります。
「クエリ」結果の場所にS3バケット名を指定します。
他の設定は適宜指定をしてください。後で変更することもできます。
「保存」をクリックするとクエリが実行できるようになります。

4.データベースの設定
Athena内にVPC Flow Logsを分析するためのデータベースを作成します。
データベース名はわかりやすいように「vpcflowlogs」としてます。
クエリエディタ画面に下記のコードをペーストして「実行」します。

CREATE DATABASE vpcflowlogs;

以下の画像はクエリエディタにコマンドをペーストした状態

コマンドが正常に実行できれば、データベース欄のプルダウンメニューに作成した「vpcflowlogs」が表示されます。

5.テーブル作成
続いてテーブル作成をします。
テーブル作成用のコマンドはAWSドキュメントに記載のコマンドを参考に実行します。
参考：AWSドキュメント

下から2行目、「LOCATION」部分の編集が必要です。VPC Flow Logs用に作成したS3バケットを指定します。

CREATE EXTERNAL TABLE IF NOT EXISTS `vpc_flow_logs` (
  `version` int, 
  `account_id` string, 
  `interface_id` string, 
  `srcaddr` string, 
  `dstaddr` string, 
  `srcport` int, 
  `dstport` int, 
  `protocol` bigint, 
  `packets` bigint, 
  `bytes` bigint, 
  `start` bigint, 
  `end` bigint, 
  `action` string, 
  `log_status` string, 
  `vpc_id` string, 
  `subnet_id` string, 
  `instance_id` string, 
  `tcp_flags` int, 
  `type` string, 
  `pkt_srcaddr` string, 
  `pkt_dstaddr` string, 
  `region` string, 
  `az_id` string, 
  `sublocation_type` string, 
  `sublocation_id` string, 
  `pkt_src_aws_service` string, 
  `pkt_dst_aws_service` string, 
  `flow_direction` string, 
  `traffic_path` int 
)
PARTITIONED BY (`date` date)
ROW FORMAT DELIMITED
FIELDS TERMINATED BY ' '
LOCATION 's3://<VPCFlowLogs用に作成したS3バケット名>/AWSLogs/{account_id}/vpcflowlogs/{region_code}/'
TBLPROPERTIES ("skip.header.line.count"="1");

コマンドが成功すると、左側の画面のテーブル欄に「vpc_flow_logs」テーブルが表示されるようになります。

6.パーティション作成
テーブルはできたもののこの時点では、データが入っていないため、まだクエリをかけることができません。パーティションを作成してデータを読み取れるようにします。パーティション作成のコマンドはAWSドキュメントに記載のコマンドを実行します。
参考：AWSドキュメント

2行目、YYYY-MM-dd部分を分析したいログの日付を指定します。
3行目、「LOCATION」部分の編集が必要です。VPC Flow Logs用に作成したS3バケットを指定します。

ALTER TABLE vpc_flow_logs
ADD PARTITION (`date`='YYYY-MM-dd')
LOCATION 's3://<VPCFlowLogs用に作成したS3バケット名>/AWSLogs/{account_id}/vpcflowlogs/{region_code}/YYYY/MM/dd';

※今回は指定日に対しての単一のパーティション作成となります。
日付が変わる度に連続してパーティション化をするには公式ドキュメントなどを参考にしてください。

これで準備は終了です。

ログ調査

フローログにたいしてクエリを実行していくことができるようになりました。参考までにいくつか一般的なクエリを紹介します。適宜、クエリエディタにコマンドをペーストして実行してみてください。

特定のポート番号（例えば22番ポート）のトラフィックを取得する例:

SELECT * FROM "vpc_flow_logs" WHERE dstport = 22

結果

"2","123456789012","eni-0123","10.0.0.1","172.16.0.1","46176","22","6","1","44","1683849324","1683849337","REJECT","OK",,,,,,,,,,,,,,,,"2023-05-11"
"2","123456789012","eni-0246","10.0.0.3","172.16.0.11","50052","22","6","1","40","1683842899","1683842958","REJECT","OK",,,,,,,,,,,,,,,,"2023-05-11"

特定のアクション（例えばREJECT）のトラフィックを取得する例:

SELECT * FROM "vpc_flow_logs" WHERE action = 'REJECT';

結果

"2","0123456789012","eni-018","198.235.**.**","172.16.0.4","51333","9418","6","1","44","1683833693","1683833722","REJECT","OK",,,,,,,,,,,,,,,,"2023-05-11"
"2","0123456789012","eni-036","35.203.**.**","172.16.0.8","55891","53306","6","1","44","1683833693","1683833722","REJECT","OK",,,,,,,,,,,,,,,,"2023-05-11"

拒否(REJECT)された送信元アドレスの上位10件を取得する例：

SELECT srcaddr, COUNT(*) AS num_of_access
FROM vpc_flow_logs
WHERE action='REJECT'
GROUP BY srcaddr
ORDER BY num_of_access DESC
LIMIT 10;

結果

#	srcaddr	        num_of_access
1	89.248.***.***	476
2	89.248.***.***	453
3	104.152.***.**	357
4	45.155.***.**	256
5	172.245.***.**	146
6	44.213.***.**	140
7	162.142.***.**	106
8	162.142.***.**	103
9	162.142.***.**	100
10	183.136.***.**	99

このようにAthenaを使えば簡単にフローログの分析をすることができるようになります。

まとめ

VPC Flow Logsを有効にすることで、ネットワークのトラフィックに関する情報を収集することができます。万が一、ネットワーク攻撃を受けた際の分析、ネットワーク疎通に関する問題解析などに利用することができます。
ぜひ有効化したいものですね。