速读摘要:主流云 WAF 都提供 REST API,可实现域名管理、规则下发、IP 名单同步、日志拉取、报表查询和告警 Webhook 对接。自动化的最大价值是打通"告警 → 分析 → 封禁 → 归档"的闭环。本文讲清 API 的典型能力和一个完整的自动化实战示例。

一、为什么需要 API
手动点控制台只能应对小规模、低频次的运维。当你有几十个域名、每天收到上百条告警、还要和自有系统打通时,就必须靠 API 把重复工作自动化。
二、WAF API 的典型能力
| 能力类别 | 具体接口 | 典型用途 |
|---|---|---|
| 域名管理 | 增删改查域名 | 批量接入、动态上下线 |
| 规则管理 | 规则增删改查、启停 | 批量下发策略 |
| IP 名单 | 黑白名单增删查 | 自动同步封禁列表 |
| 日志 | 按条件拉取日志 | 对接 SIEM、自建分析 |
| 报表 | 查询统计数据 | 自动生成日报周报 |
| 告警 | Webhook 配置 | 事件推送到自有系统 |
| 证书 | 证书上传/查询 | 自动化证书续期 |
三、一个完整的自动化闭环
最有价值的场景是:把”发现攻击”到”封禁攻击者”的全过程自动化。
自动封禁流程:
① WAF 检测到某 IP 高频攻击
↓
② 通过 Webhook 推送到你的服务
↓
③ 脚本判断:该 IP 是否在 5 分钟内被拦 > 100 次?
↓ 是
④ 调用 WAF API:把该 IP 加入黑名单(封禁 24 小时)
↓
⑤ 同步写入源站 iptables / Nginx deny 规则
↓
⑥ 推送到企业微信群通知
↓
⑦ 记录到工单系统留痕
四、代码示例
4.1 调用 WAF API 封禁 IP
import requests
API_BASE = "https://waf.example.com/api/v1"
TOKEN = "your_api_token"
def block_ip(ip, hours=24):
"""调用 WAF API 将 IP 加入黑名单"""
url = f"{API_BASE}/blacklist"
headers = {
"Authorization": f"Bearer {TOKEN}",
"Content-Type": "application/json",
}
payload = {
"ip": ip,
"action": "block",
"duration": hours * 3600,
"remark": "自动封禁:高频攻击"
}
r = requests.post(url, json=payload, headers=headers, timeout=10)
return r.json()
if __name__ == "__main__":
print(block_ip("45.123.45.67"))
4.2 处理 Webhook 告警
from flask import Flask, request
app = Flask(__name__)
@app.route("/waf-webhook", methods=["POST"])
def waf_webhook():
"""接收 WAF 告警,自动判断并封禁"""
data = request.json
src_ip = data.get("src_ip")
hit_count = data.get("hit_count", 0)
# 5 分钟内被拦超 100 次 → 自动封禁
if hit_count > 100:
block_ip(src_ip, hours=24)
notify_wechat(f"已自动封禁攻击 IP:{src_ip}")
return {"ok": True}
if __name__ == "__main__":
app.run(port=8000)
4.3 同步拉取日志做分析
import requests, json
from datetime import datetime, timedelta
def fetch_logs(hours=1):
"""拉取最近 N 小时的攻击日志"""
end = datetime.now()
start = end - timedelta(hours=hours)
url = f"{API_BASE}/logs"
params = {
"start": start.isoformat(),
"end": end.isoformat(),
"action": "block",
"limit": 1000,
}
r = requests.get(url, params=params,
headers={"Authorization": f"Bearer {TOKEN}"})
return r.json()
# 统计 Top 攻击源
logs = fetch_logs(24)
from collections import Counter
ip_count = Counter(item["src_ip"] for item in logs["items"])
for ip, cnt in ip_count.most_common(10):
print(ip, cnt)
五、自动化能覆盖的场景
| 场景 | 触发条件 | 自动动作 |
|---|---|---|
| 自动封禁 | IP 高频攻击 | 加入黑名单 |
| 自动限速 | 某 URI 请求激增 | 下发限速规则 |
| 自动扩容告警 | 流量超阈值 | 通知 + 扩容 |
| 自动日报 | 每天 8:00 | 拉报表生成简报 |
| 证书续期 | 证书剩余 < 15 天 | 自动申请部署 |
| 规则同步 | 模板更新 | 批量下发到各域名 |
六、安全注意事项
- API Token 妥善保管,不要硬编码在代码里,用环境变量或密钥管理服务
- Token 权限最小化,自动化脚本只给需要的权限
- API 调用记录日志,便于审计
- 自动封禁设置”封禁时长上限”,避免误封造成长期影响
- 关键操作(如关闭防护)不要自动化,保留人工确认
七、小结
WAF API 是把安全运营从”人力驱动”变成”系统驱动”的关键。最有价值的自动化是”告警 → 分析 → 封禁 → 归档”的闭环,能极大缩短响应时间。使用时注意 Token 安全和权限最小化,关键破坏性操作保留人工确认。
