在分析存储在 CloudWatch 或 Amazon S3 中的 AWS WAF 日志时,我有哪些选项?

上次更新日期:2022 年 7 月 11 日

我将 AWS WAF 日志存储在 Amazon CloudWatch、Amazon Simple Storage Solution (Amazon S3) 或 Amazon S3 中,作为 Amazon Kinesis Data Firehose 传输流的目标。我在分析 AWS WAF 访问日志时有哪些选项?

解决方法

要分析和筛选特定的日志请求,请为 CloudWatch 日志使用 Amazon CloudWatch Logs Insights 或为 Amazon S3 日志使用 Amazon Athena。

使用 CloudWatch Logs Insights 分析 AWS WAF 访问日志

  1. 打开 Amazon CloudWatch 控制台
  2. 在导航窗格中,选择 Logs(日志),然后选择 Log Insights
  3. 对于 Select log group(s)(选择日志组),选择一个或多个由 AWS WAF 访问日志组成的要查询的日志组。
  4. (可选)选择要查询的时段的时间范围。
  5. 使用查询语法设计查询。
  6. 选择 Run(运行)以查看日志组的结果。

以下是示例查询,您可以使用这些查询筛选 CloudWatch Logs Insights 的特定信息:

筛选特定字符串

运行此查询以根据特定字符串筛选日志:
注意:将字符串 {jndi:ldap. 替换为要搜索的字符串。

fields terminatingRuleId as Rule, action, httpRequest.country as Country, httpRequest.clientIp as ClientIP, httpRequest.httpMethod as Method,httpRequest.uri as URI
| parse @message /\{"name":"[Hh]ost\",\"value":\"(?<Host>[^"}]*)/
| parse @message /\{"name":"[Uu]ser\-[Aa]gent\",\"value\"\:\"(?<UserAgent>[^"}]*)/
| filter @message like "{jndi:ldap"
| sort action, URI desc

按主机筛选

运行此查询以按主机筛选:
注意:将主机值 www.example.com 替换为要搜索的主机。

parse @message /\{"name":"[Hh]ost\",\"value":\"(?<Host>[^"}]*)/
| filter Host = "www.example.com"
| fields terminatingRuleId as Rule, action, httpRequest.country as Country, httpRequest.clientIp as ClientIP, httpRequest.uri as URI

筛选 POST 请求

运行此查询以隔离任何 POST 请求:

parse @message /\{"name":"[Uu]ser\-[Aa]gent\",\"value\"\:\"(?<UserAgent>[^"}]*)/
| parse @message /\{"name":"[Hh]ost\",\"value":\"(?<Host>[^"}]*)/
| fields terminatingRuleId as Rule, action, httpRequest.country as Country, httpRequest.clientIp as ClientIP, httpRequest.httpMethod as Method, httpRequest.uri as URI, httpRequest.requestId as RequestID
| filter httpRequest.httpMethod ="POST"
| display Rule, action, Country, ClientIP, Method, URI, Host, UserAgent, RequestID
| sort Rule, action desc

筛选 UserAgent

运行此查询以按 UserAgent 进行筛选:
注意:User-Agent-Value 替换为您的 UserAgent

parse @message /\{"name":"[Uu]ser\-[Aa]gent\",\"value\"\:\"(?<UserAgent>[^"}]*)/
| filter UserAgent like "<User-Agent-Value>"
| fields terminatingRuleId as Rule, action, httpRequest.country as Country, httpRequest.clientIp as ClientIP, httpRequest.uri as URI

筛选不是来自某个国家/地区的请求

运行此查询以筛选并非来自特定国家/地区的请求:

fields terminatingRuleId as Rule, action, httpRequest.country as Country, httpRequest.clientIp as ClientIP, httpRequest.uri as URI
| parse @message /\{"name":"[Hh]ost\",\"value":\"(?<Host>[^"}]*)/
| parse @message /\{"name":"[Uu]ser\-[Aa]gent\",\"value\"\:\"(?<UserAgent>[^"}]*)/
| filter Country != "US"
| sort Country, action desc

筛选跨站点脚本或 SQL 注入

运行此查询以筛选跨站点脚本或 SQL 注入:

fields @timestamp, terminatingRuleId, action, httpRequest.clientIp as ClientIP, httpRequest.country as Country, terminatingRuleMatchDetails.0.conditionType as ConditionType, terminatingRuleMatchDetails.0.location as Location, terminatingRuleMatchDetails.0.matchedData.0 as MatchedData
| filter ConditionType in["XSS","SQL_INJECTION"]

基于终止规则的时间序列

运行此查询以根据终止规则筛选时间序列:

#Time Series by Terminating Rule
filter terminatingRuleId = "AWS-AWSManagedRulesCommonRuleSet"
| stats count(*) as requestCount by bin(30m)

按客户端 IP、国家/地区、URI 和规则汇总已阻止的请求

运行此查询,按客户端 IP、国家/地区、URI 和规则汇总已阻止的请求:

fields httpRequest.clientIp as ClientIP, httpRequest.country as Country, httpRequest.uri as URI, terminatingRuleId as Rule
| filter action = "BLOCK"
| stats count(*) as RequestCount by Country, ClientIP, URI, Rule
| sort RequestCount desc

热门客户端 IP

运行此查询以统计热门客户端 IP:

stats count(*) as RequestCount by httpRequest.clientIp as ClientIP
| sort RequestCount desc

热门国家/地区

运行此查询以统计热门国家/地区:

stats count(*) as RequestCount by httpRequest.country as Country
| sort RequestCount desc

热门主机

运行此查询以统计热门主机:

parse @message /\{"name":"[Hh]ost\",\"value":\"(?<Host>[^"}]*)/
| stats count(*) as RequestCount by Host
| sort RequestCount desc

热门方法

运行此查询以统计热门方法:

stats count(*)as RequestCount by httpRequest.httpMethod as Method
| sort RequestCount desc

热门终止规则

运行此查询以统计热门的终止规则:

stats count(*) as RequestCount by terminatingRuleId
| sort RequestCount desc

热门 UserAgents

运行此查询以统计热门 UserAgents

parse @message /\{"name":"[Uu]ser\-[Aa]gent\",\"value\"\:\"(?<UserAgent>[^"}]*)/
| stats count(*) as RequestCount by UserAgent
| sort RequestCount desc

未被 Default_Action 或具有 ALLOW 操作的规则终止的请求

运行此查询以筛选未被 Default_Action 或具有 ALLOW 操作的规则终止的请求:

fields @timestamp, terminatingRuleId, action, @message
| filter terminatingRuleId != 'Default_Action' and action != 'ALLOW'
| sort @timestamp desc

使用无效 Captcha 令牌的请求

运行此查询以筛选使用无效 Captcha 令牌的请求:

fields @timestamp, httpRequest.clientIp, httpRequest.requestId, captchaResponse.failureReason, @message
|filter captchaResponse.failureReason ='TOKEN_MISSING'
| sort @timestamp desc

通过基于速率的规则阻止的请求

运行此查询以筛选通过基于速率的规则阻止的请求:

fields @timestamp, httpRequest.clientIp, terminatingRuleId, httpRequest.country,@message
| filter terminatingRuleType ="RATE_BASED" ## and webaclId = "arn:aws:wafv2:us-east-1:xxxxxxxx:regional/webacl/waf-test/abcdefghijkl" ## uncomment to filter for specific WebACL
| sort requestCount desc

筛选 AWS Bot Control (ABC) 检测到的所有请求

运行此查询以筛选 ABC 检测到的所有请求:

fields @timestamp, @message
|filter @message like 'awswaf:managed:aws:bot-control'
| parse @message '"labels":[*]' as Labels
| sort @timestamp desc

使用 Amazon Athena 分析 AWS WAF 访问日志

您可以直接在 Amazon S3 存储桶中启用 AWS WAF 访问日志记录。或者,您可以使用 Amazon Kinesis Data Firehose 传输流将 AWS WAF 访问日志传输到 Amazon S3 存储桶。要在 Amazon S3 中存储日志,请参阅如何配置 AWS WAF 综合日志记录才能将日志存储在 Amazon S3 中?

当访问日志位于 Amazon S3 存储桶中时,创建 AWS WAF 表以使用 Amazon Athena 查询日志和筛选各种详细信息。

这些查询是可用于通过 Athena 查询 AWS WAF 日志的示例:

使用 AWS WAF 规则信息阻止的请求

运行此 Athena 查询,列出所有使用 AWS WAF 规则阻止的请求:

SELECT timestamp,
    action,
    httpsourcename,
    httpsourceid,
    httprequest.requestID,
    httprequest.clientip,
    webaclid,
    terminatingruleid,
    terminatingruletype,
    rulegrouplist,
    terminatingrulematchdetails
FROM "wafv2"."waf_logs"
WHERE ("action" LIKE 'BLOCK')

请求用户代理

运行此 Athena 查询以请求用户代理
注意:User-Agent 替换为您的 UserAgent 值。

select n.value, count(n.value) as count
from waf_logs
cross join
unnest(
  cast(
    httprequest.headers as ARRAY(ROW(name VARCHAR, value VARCHAR))
    )
  ) as x(n)
where n.name = 'User-Agent'
group by n.value
ORDER BY count(n.value) DESC

请求 URI

运行此 Athena 查询以检查请求 URI:

SELECT
"httprequest"."uri"
, "count"(*) "count"
FROM
  waf_logs
WHERE ("action" LIKE 'BLOCK')
GROUP BY "httprequest"."uri"
ORDER BY "count" DESC

根据 ClientIP 对被阻止的请求进行计数

运行此 Athena 查询,查看基于 ClientIP 和国家/地区的已阻止请求计数:

SELECT
  "httprequest"."clientip"
, "count"(*) "count"
, "httprequest"."country"
FROM
waf_logs
WHERE ("action" LIKE 'BLOCK')
GROUP BY "httprequest"."clientip", "httprequest"."country"
ORDER BY "count" DESC

查看请求计数

运行此 Athena 查询以查看请求计数:

SELECT 
  "httprequest"."clientip"
, "count"(*) "count"
,"httprequest"."country"
FROM
 waf_logs
WHERE ("action" LIKE
'BLOCK')
GROUP BY
"httprequest"."clientip", "httprequest"."country"
ORDER BY "count" DESC

有关其他 Athena 查询示例,请参阅AWS WAF 日志的查询示例


这篇文章对您有帮助吗?


您是否需要账单或技术支持?