通过http头搜索Athena AWS WAF日志

Question

我已经将AWS WAF Logging设置为S3，并按照文档中的描述创建了一个Athena表(https://docs.aws.amazon.com/athena/latest/ug/waf-logs.html(

但是，http标头存储为array<struct<name: string, value: string>>，而不是映射(出于正当原因(。我想写一个类似的查询

select headers.user-agent, headers.if-none-match from waf_logs where something;

如果没有匹配，则可能存在或可能不存在于标头列表中。

使用CROSS JOIN UNNEST(httprequest.headers)不起作用，因为这将创建多行。使用映射表示法不起作用，因为它是一个结构数组，而不是映射。

互联网上有很多关于如何设置表的页面，但没有那么多真实查询的示例，我也找不到关于如何通过嵌套属性进行查询的页面。

如果有任何建议，我将不胜感激。谢谢

Answer 1

下面是一个WAF查询，它应该可以处理请求头：

WITH waf_data AS (
SELECT
waf.action as action,
waf.httprequest.clientip as clientip,
waf.httprequest.country as country,
map_agg(f.name, f.value) AS kv
FROM "waf_logs" waf,
UNNEST(waf.httprequest.headers) AS t(f)
GROUP BY 1, 2, 3 
)
SELECT
waf_data.action,
waf_data.clientip,
waf_data.country,
waf_data.kv['Host'] AS host,
waf_data.kv['User-Agent'] as UserAgent,
waf_data.kv['Cookie'] as cookie
FROM waf_data
WHERE waf_data.kv['Host'] like 'waf_alb.us-east-2.elb.amazonaws.com'
LIMIT 10;

Answer 2

我使用以下内容提取HTTP头值(此处为按名称列出的Host字段(：

SELECT action, header.value as hostname, clientip, timestamp
from (
SELECT 
httprequest.clientip as clientip, 
action,
timestamp, 
httprequest.headers as headers 
FROM waf_logs 
) 
cross join unnest(headers) as c(header)
where lower(header.name) = 'host'

我用这篇文章提取了waf日志数据库中格式类似array(row("name" varchar,"value" varchar))的行数组

Answer 3

如果可能的话，我强烈建议所有开发人员不要使用UNNEST，因为大多数时候你不想要笛卡尔式的产品，而且你绝对应该避免多次不测试，因为行数会高得难以想象。只需投射httprequest.headers，然后使用过滤器提取值。

仅仅为了得到值而取消测试然后再次分组也是乏味和不稳定的，如果你需要通过键从httprequest.headers中提取多个值，你根本无法轻松稳定地得到100%正确的结果。

请记住，大多数开发人员认为他们对SQL很了解并掌握了它，只有少数人意识到要把它做好是多么困难，SQL是多么容易返回一个虚幻而令人信服的结果，让你认为你写了正确的SQL并得到了正确的结果。

WITH
targets AS (
SELECT
*,
cast(
httprequest.headers as ARRAY(ROW(name VARCHAR, value VARCHAR))
) as headers
FROM your_app.waf_logs
WHERE
"action" NOT IN ('BLOCK', 'ALLOW')
AND to_iso8601(from_unixtime(timestamp / 1000)) >= '2023-04-18'
),
result AS (
SELECT
to_iso8601(from_unixtime(timestamp / 1000)) as time_ISO_8601,
terminatingruleid,
labels,
try(
filter(
headers,
x -> LOWER(x.name) = 'user-agent'
)[1].value
) AS UserAgent,
try(
filter(
headers,
x -> LOWER(x.name) = 'x-forwarded-for'
)[1].value
) AS XForwardedForIP
FROM targets
)
SELECT
*
FROM result
;