AWS WAFログのruleGroupList内にあるCOUNTをAthenaでクエリする
2025.02.28こんにちは、なおにしです。
AWS WAFのCOUNTアクションが適用されたログをAthenaで抽出する機会がありましたのでご紹介します。
はじめに
AWS WAFのWebACLに新しいルールまたはルールグループを追加する場合、まずはCOUNTモードで適用して様子を見るという対応を取ることが多いかと思います。また、特定のアクセスがWAFによって阻害されていることが疑われる場合、例えばBot Control ルールグループ内の特定のルールのみCOUNTモードに変更することもあるかと思います。
上記を適用した場合、各ルールでCOUNTに合致したリクエスト数だけであればCloudWatch メトリクスから確認可能ですが、具体的にどのIPアドレスでどのようなHTTPリクエストがきているのかを確認するためにはログ調査が必要となります。
ログ調査をする際、例えばALLOWやBLOCKのリクエスト内容を確認するのであれば、ログ内のactionフィールドを条件に使用することが可能です。一方で、COUNTについてはactionフィールドに含まれていません。
ログ出力内でCOUNTが記録される箇所については以下の記事をご参照ください。
上記記事より、COUNTで記録されたログを抽出するためには主に以下2つのパターンを調査する必要があります。
- 【パターン1】WAF ログ直下の "nonTerminatingMatchingRules" フィールドにある "action": "COUNT"を抽出
- カスタムルールのアクションを Count に設定した場合
- ルールグループのアクションを Count にオーバーライドした場合
- 【パターン2】"ruleGroupList" フィールド内の "nonTerminatingMatchingRules" フィールドにある "action": "COUNT" を抽出
- ルールグループ内、個別のルールのデフォルトアクションを Count に設定した場合
- ルールグループ内、個別のルールのアクションを Count にオーバライドした場合
上記を実際にAthenaを使用して抽出してみます。
やってみた
事前準備
リクエスト数が多いサイトではWAFログも大量に出力されるため、Athenaを用いて分析する際にはPartition Projection (パーティション射影)を適用したテーブル作成を行う方がコスト・パフォーマンスの面で推奨となります。詳細は以下の記事もご参照ください。
今回は実際に上記記事に記載のテーブル作成用クエリを使用してPartition Projectionを適用したテーブルを作成しましたので、以降の操作・クエリは左記が前提であることをご留意ください。
【パターン1】WAF ログ直下の "nonTerminatingMatchingRules" フィールドにある "action": "COUNT"を抽出
例として、「SQL データベースマネージドルールグループ(AWSManagedRulesSQLiRuleSet)」をCOUNTに設定している場合に発生したリクエスト内容を確認してみます。
以下のようCloudWatchメトリクスで確認した2件のリクエスト内容を確認したいという状況だったとします。
WebACLで設定するルールグループ名も「AWSManagedRulesSQLiRuleSet」とした場合、Athenaで実行するクエリは以下のとおりです。
WITH
waf_logs_partition AS (
SELECT
*
FROM
waf_logs
WHERE
date BETWEEN format_datetime(from_iso8601_timestamp('2025-02-16T05:00:00+09:00') AT TIME ZONE 'UTC', 'yyyy/MM/dd/HH') AND
format_datetime(from_iso8601_timestamp('2025-02-16T06:00:00+09:00') AT TIME ZONE 'UTC', 'yyyy/MM/dd/HH')
)
SELECT
date_format(from_unixtime(timestamp/1000) AT TIME ZONE 'Asia/Tokyo', '%Y/%m/%d %H:%i:%s') AS timestamp_converted_jst,
nonTermRule,
httprequest
FROM
waf_logs_partition, UNNEST(nonTerminatingMatchingRules) t(nonTermRule)
WHERE
timestamp BETWEEN (to_unixtime(from_iso8601_timestamp('2025-02-16T05:04:00+09:00')) * 1000) AND
(to_unixtime(from_iso8601_timestamp('2025-02-16T05:10:00+09:00')) * 1000)
AND nonTermRule.action = 'COUNT'
AND nonTermRule.ruleid = 'AWSManagedRulesSQLiRuleSet'
ORDER BY
timestamp DESC
実行結果は以下のように出力されました。
| # | timestamp_converted_jst | nonTermRule | httprequest |
|---|---|---|---|
| 1 | 2025/02/16 05:06:11 | {ruleid=AWSManagedRulesSQLiRuleSet, action=COUNT, rulematchdetails=[{conditiontype=SQL_INJECTION, sensitivitylevel=LOW, location=ALL_QUERY_ARGS, matcheddata=[2280, ), as, tempxtestxtable, where]}], challengeresponse=null, captcharesponse=null} | {clientip=xxx.xxx.xxx.xxx, country=US, headers=[{name=Host, value=example.com}, {name=User-Agent, value=Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/87.0.4280.40 Safari/537.36 Edg/87.0.664.24}, {name=Accept-Encoding, value=gzip, deflate}, {name=Accept, value=/}, {name=Connection, value=keep-alive}], uri=/test, httpversion=HTTP/1.1, httpmethod=GET, requestid=p0opOwcRzsmRLs4R9XedzliB8cnLHdut-xzxcIwHnaPCej1-9Mo4TQ==} |
| 2 | 2025/02/16 05:05:02 | {ruleid=AWSManagedRulesSQLiRuleSet, action=COUNT, rulematchdetails=[{conditiontype=SQL_INJECTION, sensitivitylevel=LOW, location=ALL_QUERY_ARGS, matcheddata=[2280, ), as, tempxtestxtable, where]}], challengeresponse=null, captcharesponse=null} | {clientip=yyy.yyy.yyy.yyy, country=US, headers=[{name=Host, value=example.com}, {name=User-Agent, value=Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/84.0.4280.40 Safari/537.36 Edg/87.0.664.24}, {name=Accept-Encoding, value=gzip, deflate}, {name=Accept, value=/}, {name=Connection, value=keep-alive}], uri=/test, httpversion=HTTP/1.1, httpmethod=GET, requestid=xh1ZROOA252iqb36o5dN3_ReVeHV3qVfwDCm4EpnbSZL7-4YE6WfTw==} |
実行するクエリについて補足します。
まず、以下のWITH句によって副問合せのように一時テーブル作成しています。この一時テーブルでは、Partition Projectionの設定で追加したdateフィールドに対して時間の範囲指定を行なっています。
CloudWatchメトリクスではおそらくローカルタイムゾーンを指定してモニタリングすることが多いかと思いますので、そのまま時刻をJSTで指定できるようにしています。
WITH
waf_logs_partition AS (
SELECT
*
FROM
waf_logs
WHERE
date BETWEEN format_datetime(from_iso8601_timestamp('2025-02-16T05:00:00+09:00') AT TIME ZONE 'UTC', 'yyyy/MM/dd/HH') AND
format_datetime(from_iso8601_timestamp('2025-02-16T06:00:00+09:00') AT TIME ZONE 'UTC', 'yyyy/MM/dd/HH')
)
作成された一時テーブル(waf_logs_partition)に対して、以下のクエリが実行されている形になります。
「UNNEST(nonterminatingmatchingrules) t(nonTermRule)」によって、配列形式の「nonterminatingmatchingrules」を「nonTermRule」として展開することで、「nonTermRule.action」や「nonTermRule.ruleid」を指定できるようになっています。
また、一時テーブルを作成する時の時間帯指定では、パーティションを作成する単位を1時間ごとにしていることから、1時間単位の粒度による指定となっています。このため、より詳細な時間指定は以下のクエリ内でtimestampフィールドを使って改めて行なっています。
SELECT
date_format(from_unixtime(timestamp/1000) AT TIME ZONE 'Asia/Tokyo', '%Y/%m/%d %H:%i:%s') AS timestamp_converted_jst,
nonTermRule,
httprequest
FROM
waf_logs_partition, UNNEST(nonterminatingmatchingrules) t(nonTermRule)
WHERE
timestamp BETWEEN (to_unixtime(from_iso8601_timestamp('2025-02-16T05:04:00+09:00')) * 1000) AND
(to_unixtime(from_iso8601_timestamp('2025-02-16T05:10:00+09:00')) * 1000)
AND nonTermRule.action = 'COUNT'
AND nonTermRule.ruleid = 'AWSManagedRulesSQLiRuleSet'
ORDER BY
timestamp DESC
【パターン2】"ruleGroupList" フィールド内の "nonTerminatingMatchingRules" フィールドにある "action": "COUNT" を抽出
例として、「AWS WAF Bot Control ルールグループ(AWSManagedRulesBotControlRuleSet)」内のルール「TGT_ML_CoordinatedActivityHigh」にでCOUNTが記録されたリクエスト内容を確認してみます。
「TGT_ML_CoordinatedActivityHigh」はBot Control ルールグループのバージョン2.0まではCOUNTとしてルールのアクションが適用されていましたが、バージョン3.0以降はCAPTCHAになっています。このようなルールグループのバージョン選択については以下の記事もご参照ください。
以下のようCloudWatchメトリクスで確認した1件のリクエスト内容を確認したいという状況だったとします。
WebACLで設定するルールグループ名も「AWSManagedRulesBotControlRuleSet」とした場合、Athenaで実行するクエリは以下のとおりです。
WITH
waf_logs_partition AS (
SELECT
*
FROM
waf_logs, UNNEST(rulegrouplist) t(rulegroup)
WHERE
date BETWEEN format_datetime(from_iso8601_timestamp('2025-02-16T05:00:00+09:00') AT TIME ZONE 'UTC', 'yyyy/MM/dd/HH') AND
format_datetime(from_iso8601_timestamp('2025-02-16T06:00:00+09:00') AT TIME ZONE 'UTC', 'yyyy/MM/dd/HH')
AND cardinality(rulegroup.nonterminatingmatchingrules) > 0
)
SELECT
date_format(from_unixtime(timestamp/1000) AT TIME ZONE 'Asia/Tokyo', '%Y/%m/%d %H:%i:%s') AS timestamp_converted_jst,
nonterminatingmatchingrule,
httprequest
FROM
waf_logs_partition, UNNEST(rulegroup.nonterminatingmatchingrules) t(nonterminatingmatchingrule)
WHERE
timestamp BETWEEN (to_unixtime(from_iso8601_timestamp('2025-02-16T05:04:00+09:00')) * 1000) AND
(to_unixtime(from_iso8601_timestamp('2025-02-16T05:10:00+09:00')) * 1000)
AND nonterminatingmatchingrule.action = 'COUNT'
AND nonterminatingmatchingrule.ruleid = 'TGT_ML_CoordinatedActivityHigh'
ORDER BY
timestamp DESC
実行結果は以下のように出力されました。
| # | timestamp_converted_jst | nonterminatingmatchingrule | httprequest |
|---|---|---|---|
| 1 | 2025/02/16 05:08:54 | {ruleid=TGT_ML_CoordinatedActivityHigh, action=COUNT, overriddenaction=null, rulematchdetails=[], challengeresponse=null, captcharesponse=null} | {clientip=zzz.zzz.zzz.zzz, country=LV, headers=[{name=Accept, value=text/html, application/xhtml+xml; q=0.9}, {name=User-Agent, value=Mozilla/5.0 AppleWebKit/537.36 (KHTML, like Gecko; compatible; GeedoProductSearch;) Chrome/7 8.0.3945.88 Safari/537.36}, {name=Host, value=example.com}, {name=Accept-Encoding, value=gzip, deflate, br}], uri=/test, args=, httpversion=HTTP/1.1, httpmethod=GET, requestid=adww0kFS346cuiv4CmgjAI3oALrU1c3A3b5oUkaEDIA6fXAI0cY4qQ==} |
実行するクエリについて補足します。
基本的にはパターン1と同じ構成ですが、今回は以下の一時テーブルを作成する以下の部分で、「cardinality(rulegroup.nonterminatingmatchingrules) > 0」を指定しています。
WITH
waf_logs_partition AS (
SELECT
*
FROM
waf_logs, UNNEST(rulegrouplist) t(rulegroup)
WHERE
date BETWEEN format_datetime(from_iso8601_timestamp('2025-02-16T05:00:00+09:00') AT TIME ZONE 'UTC', 'yyyy/MM/dd/HH') AND
format_datetime(from_iso8601_timestamp('2025-02-16T06:00:00+09:00') AT TIME ZONE 'UTC', 'yyyy/MM/dd/HH')
AND cardinality(rulegroup.nonterminatingmatchingrules) > 0
)
「nonterminatingmatchingrules」はリクエストに一致した非終了型ルールのリストですが、このフィールドは終了アクションによるログの場合は中身が空になります。nonterminatingmatchingrulesが空の状態で後続の処理においてUNNESTを試行するとエラーになるため、cardinalityを使用してnonterminatingmatchingrulesの要素数が0より大きいことも抽出条件に加えています。その他の考え方はパターン1と同じであるため割愛します。
まとめ
Athenaを使用してCOUNTで記録されたWAFログを抽出してみました。実際にログ分析を実施する際はクエリを適宜変更しながら目的の情報を検索・収集する流れになるかと思います。本記事のようにAthenaを使用することでAWS上でログ分析を行うことができますが、執筆している最中に最近ではDuckDBを使用してS3に保存されたアクセスログを分析することも多くなっていることを知ったので、そちらもやってみたいと思いました。
本記事がどなたかのお役に立てれば幸いです。
