SOAR 自动化响应实战：WAF 日志联动自动封禁

作者

SOAR 自动化响应是安全运营从「人工翻日志」转向「机器闭环处置」的关键一步。多数站点的现状是：WAF 与 Nginx 日志每天都在产生海量告警，运维却只能事后拉日志查 IP，从发现到封禁往往延迟几十分钟，足够攻击者把数据库拖走。本文给出一套可落地的轻量实现：用滑动窗口计数做检测、用 Redis 保存状态、用 WAF API 与 nftables 双通道执行封禁，全套代码不超过百行，单机即可运行。

适用场景

  • 站点已有 WAF 或 Nginx 反向代理，但封禁仍靠人工在控制台点选，响应延迟超过 10 分钟。
  • 遭遇持续性 CC、目录扫描、漏洞探测，需要按 IP 自动限速与临时封禁。
  • 希望把「检测—判定—处置—回滚」固化成代码，避免交接时经验断层。
  • 预算有限，不打算采购商业 SOAR 平台，用一台 2 核 4G 的机器即可承担。

前置条件

  • Linux 主机（本文以 Debian 12 为例），已安装 Python 3.9+、Redis 6+、nftables。
  • Nginx 访问日志可改为 JSON 格式，或已有 WAF 提供的结构化日志导出。
  • WAF 开放了 IP 封禁 API，具备可长期使用的 Token；若没有 API，本文的 nftables 本地封禁通道可独立工作。
  • 已知本机监控、CDN 回源、办公出口等必须放行的 IP 段，提前整理成白名单。

原理说明

SOAR 的本质不是「更聪明的 WAF」，而是一个把处置动作编排起来的状态机。整条链路分四段：采集（日志进入管道）、检测（对每个来源 IP 累计风险分）、决策（超过阈值则分级处置）、执行（调用 WAF 接口与本地防火墙，并记录状态防止重复封禁）。

检测环节采用带权滑动窗口而不是简单的请求计数：普通 200 响应记 1 分，探测敏感路径（../、/etc/passwd、/.git/）记 5 分，404 且路径命中后台特征记 3 分。这样扫描器打几十个请求就能触顶，而正常用户浏览几百次页面也不会被误伤。窗口用 Redis 的 INCRBY 加 EXPIRE 实现，天然支持进程重启后状态不丢。

执行环节必须做双通道：WAF 侧封禁负责边界拦截，本地 nftables 集合负责兜底。两者互为备份，任一通道故障都不会让处置彻底失效。同时所有封禁都写入 Redis 的 soar:ban:<ip> 键并设置 TTL，保证脚本重复消费同一条日志时不会重复调用接口，也为后续一键解封留下依据。

操作步骤

步骤一：把访问日志改成结构化 JSON

在 /etc/nginx/nginx.conf 的 http 块中加入日志格式，并让业务站点引用它。字段越精简越好，避免日志膨胀。

log_format sec_json escape=json '{"time":"$time_iso8601","ip":"$remote_addr",'
                                '"method":"$request_method","path":"$uri",'
                                '"status":"$status","ua":"$http_user_agent",'
                                '"rt":"$request_time","host":"$host"}';

access_log /var/log/nginx/sec_access.log sec_json buffer=64k flush=5s;

执行 nginx -t && nginx -s reload 后，确认日志已经变成单行 JSON：tail -n 2 /var/log/nginx/sec_access.log。注意 escape=json 不可省略，否则 UA 里的双引号会破坏 JSON 结构，导致后续解析整行丢弃。

步骤二：初始化本地封禁集合

nftables 的集合支持带超时元素，正好对应「临时封禁」语义，不需要额外写定时清理任务。

nft add table inet filter
nft add set inet filter blackhole '{ type ipv4_addr; flags timeout; }'
nft add chain inet filter input '{ type filter hook input priority 0 ; }'
nft add rule inet filter input ip saddr @blackhole drop

# 持久化，重启后规则仍在
nft list ruleset > /etc/nftables.conf
systemctl enable --now nftables

步骤三：编写检测与执行脚本

把以下内容保存为 /opt/soar/soar.py。脚本只做一件事：跟随日志、算分、超阈值就封。

#!/usr/bin/env python3
import json, re, subprocess, urllib.request
import redis

R = redis.Redis(host="127.0.0.1", port=6379, db=2, decode_responses=True)
LOG        = "/var/log/nginx/sec_access.log"
WINDOW     = 60        # 统计窗口，秒
THRESHOLD  = 120       # 窗口内累计风险分上限
BAN_TTL    = 3600      # 封禁时长，秒
WAF_API    = "https://waf.example.com/api/v1/ip/block"
TOKEN      = open("/etc/soar/waf.token").read().strip()
WHITELIST  = {"127.0.0.1", "10.0.0.8"}   # 监控机与回源 IP

SENSITIVE = re.compile(r"(\.\./|/etc/passwd|/\.git/|/wp-config\.php|\.env)")
SCANNER   = re.compile(r"(phpmyadmin|admin\.php|wp-login\.php|\.sql$|\.bak$)")

def score(rec):
    path = rec.get("path", "")
    if SENSITIVE.search(path):
        return 5
    if rec.get("status") == "404" and SCANNER.search(path):
        return 3
    return 1

def window_add(ip, weight):
    key = "soar:cnt:" + ip
    pipe = R.pipeline()
    pipe.incrby(key, weight)
    pipe.expire(key, WINDOW)          # 每次写入刷新 TTL，滑动窗口近似
    return pipe.execute()[0]

def ban(ip, reason):
    if ip in WHITELIST or R.exists("soar:ban:" + ip):
        return
    try:
        body = json.dumps({"ip": ip, "ttl": BAN_TTL, "remark": reason}).encode()
        rq = urllib.request.Request(
            WAF_API, body,
            {"Content-Type": "application/json", "Authorization": "Bearer " + TOKEN})
        urllib.request.urlopen(rq, timeout=8).read()
    except Exception as e:
        print("[WAF-ERR]", ip, e)      # WAF 通道失败不阻断本地兜底
    subprocess.run(["nft", "add", "element", "inet", "filter", "blackhole",
                    "{ %s timeout 1h }" % ip], check=False)
    R.setex("soar:ban:" + ip, BAN_TTL, reason)
    print("[BAN]", ip, reason, flush=True)

def follow(path):
    proc = subprocess.Popen(["tail", "-F", "-n0", path],
                            stdout=subprocess.PIPE, text=True)
    for line in proc.stdout:
        try:
            rec = json.loads(line)
        except ValueError:
            continue                    # 半行或非法 JSON，直接跳过
        ip = rec.get("ip")
        if not ip:
            continue
        cnt = window_add(ip, score(rec))
        if cnt >= THRESHOLD:
            ban(ip, "risk=%d within %ds" % (cnt, WINDOW))

if __name__ == "__main__":
    follow(LOG)

关键参数说明：WINDOW 控制在 60 秒即可覆盖绝大多数突发型攻击；THRESHOLD 需要结合站点真实流量标定，日 PV 十万级以下从 120 起调比较稳妥；BAN_TTL 不宜设成永久，1 小时能让误封自愈，也不会给攻击者留出反复试探的空间。

步骤四：以 systemd 托管并限制权限

cat > /etc/systemd/system/soar.service <<'EOF'
[Unit]
Description=SOAR auto response
After=network.target redis-server.service nginx.service

[Service]
Type=simple
ExecStart=/usr/bin/python3 /opt/soar/soar.py
Restart=always
RestartSec=5
NoNewPrivileges=true
ProtectSystem=strict
ReadWritePaths=/var/log/nginx

[Install]
WantedBy=multi-user.target
EOF

systemctl daemon-reload && systemctl enable --now soar
journalctl -u soar -f

Token 文件权限设为 chmod 600 /etc/soar/waf.token 并归属 root，避免普通账号读取后绕过 WAF 的封禁接口。

步骤五：解封与审计留痕

误封不可避免，所以解封必须是一条命令而不是手动改三处状态。补一个子命令即可：

def unban(ip):
    R.delete("soar:cnt:" + ip, "soar:ban:" + ip)
    subprocess.run(["nft", "delete", "element", "inet", "filter", "blackhole",
                    "{ %s }" % ip], check=False)
    print("[UNBAN]", ip)

# 调用：python3 -c "import soar; soar.unban('203.0.113.9')"

同时把每次封禁写入一条审计日志（可直接复用 print，由 journald 收集），字段最少包含时间、IP、风险分、命中规则、执行通道。等保测评时，这份记录就是「安全事件处置留痕」的证据。

配置验证

验证分四步，任何一步不通过都不要急着上线。

  1. 日志可解析：tail -n 20 /var/log/nginx/sec_access.log | python3 -m json.tool 逐行无报错。
  2. 计数在增长：用测试 IP 请求几次后执行 redis-cli -n 2 get soar:cnt:<测试IP>，数值应为累计风险分。
  3. 封禁能触发：ab -n 300 -c 20 http://127.0.0.1/.env，观察 journalctl -u soar -f 是否打印 [BAN]，随后 nft list set inet filter blackhole 应出现该 IP 与剩余超时。
  4. 拦截已生效：curl -I -m 5 http://<站点>/ 从被封 IP 发起应超时或返回 403；解封后立即恢复 200。

另外建议跑一次误封压测：用正常 UA 连续访问首页 500 次，确认风险分始终停在个位数、不触发封禁。这一步比攻击测试更重要，因为误封真实用户的代价远高于漏封一次扫描。

常见问题

FAQ 1：上线后误封了正常用户，怎么把损失降到最低？

三层防护：一是把 CDN 回源段、搜索引擎爬虫 IP 段、支付回调地址加入 WHITELIST；二是把 THRESHOLD 与 BAN_TTL 设为可热更新（从 Redis 或配置文件读取），出问题时改值不用重启；三是先只做「告警不封禁」的观察期，跑三到七天，把日志里被判定超阈值的 IP 人工过一遍，确认没有正常业务 IP 再打开封禁开关。

FAQ 2：已经有 WAF 自带的 CC 防护，还需要自己写这套吗？

两者定位不同。WAF 的 CC 防护基于单 IP 频率与特征，属于「通用规则」；本文这套负责的是「跨设备、跨事件的关联处置」，比如同一个 /24 段内多 IP 轮换扫描、命中高危路径后立即封禁而不只是拦截单次请求。实践中的分工是：单请求级别交给 WAF，IP 级别的累计行为交给 SOAR。若 WAF 已支持自定义规则与 API 联动，也可以把本文脚本简化成只调用其接口，检测逻辑仍建议自己保留，便于按业务调整阈值。

FAQ 3：为什么用 Redis 而不是脚本内存里的字典？

三个原因：脚本重启或发版后内存状态会清零，攻击者只要卡在重启窗口就能绕过；多台前端需要共享同一个计数视图；Redis 自带 TTL，省掉手写清理逻辑。代价是多一个依赖，用 maxmemory 256mb 加 maxmemory-policy allkeys-lru 限制即可，避免计数键把内存吃满。

总结

这套方案的核心不是代码量，而是把「响应」变成可重复执行的流程：日志结构统一、计分规则可调、执行通道双备份、封禁状态可追溯、解封只需一条命令。落地顺序建议按「观察期—半自动—全自动」三阶段推进，每阶段至少跑一周并复盘误封记录。等这套闭环稳定后，再往上叠加告警推送、工单联动、攻击画像，才不会因为基础不稳而反复返工。