WAF日志之自动化AI分析
原创
pippybear
pippybear
安全无界
2026年10月13日 10:00
上海
在小说阅读器读本章
去阅读
在公众号小说中沉浸阅读
之前简单聊过一波WAF自动化封禁（基于AWS），刚好前面有时间，就抽空把内部的流程再进一步梳理完善了一下，简单聊聊思路。
最初的一版比较糙，似乎还是去年年初干的，主要是为了解决AWS自动化封禁问题，当时思路也比较简单，使用Eventbridge定时触发Lambda提取WAF Log进行分析，然后针对Rate异常的IP进行简单行为分析，然后Block。
设计思路
在这一年多里也是不断的修修补补变成了现在的版本，简单描述一下现在的版本，前面的逻辑差不多还是Eventbridge定时触发Lambda提取WAF Log，不过分析这个稍微优化了一下，主要分成3大块。
IP提取
1、规则匹配--触发直接进入block序列2、AI分析行为--未触发规则匹配的rate事件使用AI分析行为
IP威胁情报分提取
这里推荐一个apiabuseipdb.com，这个还挺不错的，DB挺全面，主要是每天的免费额度还挺多的：1000个IP。基本上中小型企业筛选攻击IP后查询基本够用。
威胁情报分这一块主要用于block判断和释放判断。
添加释放逻辑
V1版本只有封禁逻辑，但其实如果IP非持续攻击，其实也没有必要一直封禁着，所以现在的版本添加了自动释放逻辑。
代码实现
IP封禁
这一块逻辑其实很简单，不管是从Athena还是Cloudwatch抑或是S3提取都行，我这里采用的是从Athena提取（另外提一个小点，Athena的DB记得采用分区进行，如果不分区每一次读取的都是整个log，费钱！！！）
async function startAthenaQuery(database, query) { const startRes = await athena.startQueryExecution({ QueryString: query, QueryExecutionContext: { Database: database }, ResultConfiguration: { OutputLocation: ATHENA_OUTPUT } }).promise();
const queryExecutionId = startRes.QueryExecutionId;
while (true) { const statusRes = await athena.getQueryExecution({ QueryExecutionId: queryExecutionId }).promise(); const state = statusRes.QueryExecution.Status.State; if ([\"SUCCEEDED\", \"FAILED\", \"CANCELLED\"].includes(state)) break; await new Promise(r => setTimeout(r, 5000)); }
let allRows = []; let nextToken = null;
do { const res = await athena.getQueryResults({ QueryExecutionId: queryExecutionId, NextToken: nextToken }).promise();
const rows = res.ResultSet.Rows.map(r => r.Data.map(d => d.VarCharValue || \"\")); allRows = allRows.concat(rows);
nextToken = res.NextToken; } while (nextToken);
if (allRows.length > 0) { allRows = allRows.slice(1); }
return allRows;}
function multiDecode(str) { let prev = \"\"; let cur = str; let count = 0;
while (cur !== prev && count < 5) { prev = cur; try { cur = decodeURIComponent(cur); } catch (e) { break; } count++; }
return cur;}
提取Log后，对Log进行分析（先规则筛选+Rate提取分类）。
function classifyWAFLogs(rows) { const ip_logs = {};
for (const [ruleidRaw, clientip, uriRaw, argsRaw, user_agent] of rows) { if (IP_WHITELIST.includes(clientip)) { console.log(`🟦 白名单 IP 忽略：${clientip}`); continue; }
const ruleid = (ruleidRaw || \"\").toString(); const uri = (uriRaw || \"\").toString(); const args = (argsRaw || \"\").toString();
if (!ip_logs[clientip]) { ip_logs[clientip] = { ruleids: new Set(), user_agents: new Set(), records: [], category: \"unknown\", matchedIndex: null, hitRule: null }; }
ip_logs[clientip].records.push({ ruleid, uri, args, user_agent }); ip_logs[clientip].ruleids.add(ruleid); ip_logs[clientip].user_agents.add(user_agent); }
for (const [ip, data] of Object.entries(ip_logs)) { for (let i = 0; i < data.records.length; i++) { const { ruleid, uri, args } = data.records[i]; const ruleidLower = ruleid.toLowerCase(); const uriLower = (uri || \"\").toLowerCase(); const argsLower = (args || \"\").toLowerCase();
if (KNOWN_BAD_RULES.some(badRule => ruleid.toLowerCase().includes(badRule.toLowerCase()))) { data.category = \"attack\"; data.hitRule = ruleid; data.matchedIndex = i; break; }
if ( uri_suspicious_keywords.some(k => uriLower.includes(k)) || matchSuspicious(argsLower, args_suspicious_keywords) || matchSQLi(argsLower) || matchSQLi(uriLower) ) { data.category = \"attack\"; data.hitRule = \"(keyword-match)\"; data.matchedIndex = i; break; }
if (RATE_RULE_KEYWORDS.some(k => ruleidLower.includes(k))) { data.category = \"rate\"; data.hitRule = ruleid; data.matchedIndex = i; break; } } }
return ip_logs;}
然后使用AI对Rate类Log进行分析，判断是否为攻击行为。
IP威胁情报分提取
每一次首次出现的攻击IP，程序会自动对IP进行威胁情报分提取并存入DB。
let intel = null; if (!existing) { console.log(\"🔍 首次出现，查询威胁情报:\", ip); intel = await analyzeIPIntel(ip, secrets); }
提取逻辑。
async function analyzeIPIntel(ip, secrets) { try { const res = await checkIP(ip, secrets);
const d = res?.data;
if (!d) return null;
return { intelScore: Number(d.abuseConfidenceScore || 0), intelMarked: d.totalReports > 0, intelLastCheck: new Date().toISOString() }; } catch (e) { console.error(\"❌ 威胁情报查询失败:\", ip, e.message); return null; }}
function checkIP(ip, secrets) { return new Promise((resolve, reject) => { const options = { hostname: \"api.abuseipdb.com\", path: `/api/v2/check?ipAddress=${ip}&maxAgeInDays=90&verbose=true`, method: \"GET\", headers: { Key: secrets.abuseipdb_key, Accept: \"application/json\", }, };
const req = https.request(options, (res) => { let data = \"\";
res.on(\"data\", (chunk) => (data += chunk)); res.on(\"end\", () => { try { const json = JSON.parse(data); resolve(json); } catch (e) { reject(e); } }); });
req.on(\"error\", reject); req.end(); });}
释放逻辑
释放逻辑在每天一次的执行逻辑里面，具体释放大家可以根据自己的业务来做调整，我这里是设置了几个门槛：是否持续攻击persistent、是否属于高威胁情报IP。针对不同门槛释放时间不同。
async function runLifecycleJob(secrets) { console.log(\"🚀 开始IP生命周期维护（Query attack + ai-rate, triggerCount>=3）\");
const toBlock = []; const toRelease = []; const releasedSuccess = []; const blockedSuccess = []; const seen = new Set();
async function collectBySource(source) { let lastKey = null;
do { const res = await queryAttackCandidates(source, lastKey);
console.log(`📦 [${source}] 本批获取 ${res.Items?.length || 0} 条记录, LastEvaluatedKey:`, res.LastEvaluatedKey);
for (const item of (res.Items || [])) { if (!item.ip || String(item.ip).startsWith(\"__\")) continue;
const ip = item.ip; if (seen.has(ip)) continue; seen.add(ip);
const triggerCount = Number(item.triggerCount || 0); const lastSource = (item.lastSource || \"\").toLowerCase();
// lifecycle 封禁逻辑只针对 attack if (lastSource === \"attack\") { if ((item.intelScore ?? 0) >= 50 && !item.isBlocked) { console.log(\"🚫 命中威胁情报IP多次扫描自动封禁条件:\", item.ip); toBlock.push({ ip: item.ip, item, reason: \"威胁情报高危封禁\" }); }
if ((item.intelScore ?? 0) < 50 && !item.isBlocked && triggerCount >= 5) { console.log(\"🚫 命中普通IP多次扫描自动封禁条件:\", item.ip); toBlock.push({ ip: item.ip, item, reason: \"普通多次扫描封禁\" }); } }
const score = Number(item.intelScore ?? 0); const idleDays = daysSince(item.lastSeen); const hasTag = item.tag === \"persistent\";
let attackDaySetRaw = item.attackDaySet; let attackDaySet = [];
if (Array.isArray(attackDaySetRaw)) { attackDaySet = attackDaySetRaw; } else if (attackDaySetRaw && typeof attackDaySetRaw === \"object\" && Array.isArray(attackDaySetRaw.values)) { attackDaySet = attackDaySetRaw.values; } else if (typeof attackDaySetRaw === \"string\") { attackDaySet = attackDaySetRaw.split(\";\").filter(Boolean); }
const last7 = getLastNDays(7); const activeDays = attackDaySet.filter(d => last7.includes(d));
/* ---------- persistent ---------- */ if (!hasTag && activeDays.length >= 4) { console.log(`🔥 标记 persistent: ${ip}`);
await dynamo.update({ TableName: DYNAMODB_TABLE, Key: { ip }, UpdateExpression: \"SET tag = :t\", ExpressionAttributeValues: { \":t\": \"persistent\" } }).promise();
continue; }
/* ---------- 普通释放 ---------- */ if (!hasTag && !item.tagClearedAt && canReleaseByIntel(score, idleDays)) { console.log(`🟢 命中普通释放条件: ${ip}`); toRelease.push({ ip, item, type: \"普通释放\" }); continue; }
/* ---------- persistent 降级 ---------- */ if (hasTag && idleDays >= 7 && !item.tagClearedAt) { console.log(`🟡 清 persistent: ${ip}`);
await dynamo.update({ TableName: DYNAMODB_TABLE, Key: { ip }, UpdateExpression: ` REMOVE tag SET tagClearedAt = :t `, ExpressionAttributeValues: { \":t\": new Date().toISOString() } }).promise();
continue; }
/* ---------- 二次释放 ---------- */ if ( !hasTag && item.tagClearedAt && daysSince(item.tagClearedAt) >= 7 && canReleaseByIntel(score, idleDays) ) { console.log(`🟢 命中 persistent释放条件: ${ip}`); toRelease.push({ ip, item, type: \"persistent释放\" }); } }
lastKey = res.LastEvaluatedKey; } while (lastKey); }
await collectBySource(\"attack\"); await collectBySource(\"ai-rate\");
const uniqueBlockMap = new Map(); for (const b of toBlock) { if (!uniqueBlockMap.has(b.ip)) { uniqueBlockMap.set(b.ip, b); } }
const uniqueReleaseMap = new Map(); for (const r of toRelease) { if (!uniqueReleaseMap.has(r.ip)) { uniqueReleaseMap.set(r.ip, r); } }
const finalBlocks = Array.from(uniqueBlockMap.values()); const finalReleases = Array.from(uniqueReleaseMap.values());
console.log(`📌 待释放: ${finalReleases.length}`); console.log(`📌 待封禁: ${finalBlocks.length}`);
const blockIps = finalBlocks.map(x => x.ip); const releaseIps = finalReleases.map(x => x.ip);
try { if (blockIps.length) { console.log(`🚫 批量封禁 ${blockIps.length} 个 IP`); await blockIPs(blockIps); }
if (releaseIps.length) { console.log(`🟢 批量解封 ${releaseIps.length} 个 IP`); await unblockIPs(releaseIps); } } catch (err) { console.error(\"❌ 批量更新 IPSet 失败，终止后续 DB 更新:\", err); throw err; }
for (const r of finalReleases) { try { console.log(`🟢 ${r.type}: ${r.ip}`);
await dynamo.update({ TableName: DYNAMODB_TABLE, Key: { ip: r.ip }, UpdateExpression: ` REMOVE attackDaySet, tag, tagClearedAt SET should_release = :r, isBlocked = :b, triggerCount = :z, lastSeen = :n, releasedAt = :rt `, ExpressionAttributeValues: { \":r\": true, \":b\": false, \":z\": 0, \":n\": new Date().toISOString(), \":rt\": new Date().toISOString() } }).promise();
releasedSuccess.push({ ip: r.ip, type: r.type, intelScore: r.item?.intelScore, triggerCount: r.item?.triggerCount }); } catch (err) { console.error(`❌ 释放失败 ${r.ip}:`, err); } }
for (const b of finalBlocks) { try { console.log(`🚫 ${b.reason}: ${b.ip}`);
await dynamo.update({ TableName: DYNAMODB_TABLE, Key: { ip: b.ip }, UpdateExpression: \"SET isBlocked = :blocked, blockedAt = :time\", ExpressionAttributeValues: { \":blocked\": true, \":time\": new Date().toISOString() } }).promise();
blockedSuccess.push({ ip: b.ip, reason: b.reason, intelScore: b.item?.intelScore, triggerCount: b.item?.triggerCount }); } catch (err) { console.error(`❌ 封禁失败 ${b.ip}:`, err); } }
await notifyIPReleaseBatch(releasedSuccess, secrets); await notifyIPBlockBatch(blockedSuccess, secrets, \"🚫 IP 批量封禁通知\");
console.log(\"✅ 生命周期任务完成\");}
新增的几块核心代码如上，至于中间还涉及到的比如每日封禁/释放统计、如何实现DB、IPset同步操作等等逻辑代码就不展示了，实现起来其实也比较简单。直接看看最终效果。
版权声明
本站仅做备份收录，仅供研究与教学参考之用。
读者将信息用于其他用途的，全部法律及连带责任由读者自行承担，本站不承担任何责任。
评论