Cloudflare 正在向所有账户开放 Cloudforce One Threat Events Platform，并推出 Threat Signals。它要解决的不是“缺少威胁报告”，而是报告读完之后仍需人工摘录 IP、域名等指标，再逐条转换成 WAF 条件的问题。Threat Signals 会自动解析开源威胁报告、提取结构化指标，并把相关威胁上下文连接到 WAF 规则。
真正的变化是缩短情报到防护的距离
传统开源威胁情报流程通常包含多次手工转换：安全团队阅读报告，复制可疑指标，确认格式与可信度，然后交给运维或应用团队修改规则。每次转交都会增加延迟，也容易丢失来源、事件背景和判断依据。
Threat Signals 将这条链路压缩为三个动作：
- 解析公开威胁报告中的非结构化文本；
- 提取可被机器处理的结构化指标；
- 将指标及其上下文带到 WAF 规则决策中。
这里的重点不只是“识别了一个 IP”。同一个 IP 可能属于攻击基础设施，也可能是共享代理、云主机或被临时入侵的服务器。只有把指标和威胁事件背景放在一起，规则维护者才更容易决定应该记录、质询还是阻断。
所谓 agentic skill，也不应被理解为无需监督的自动封禁。更稳妥的理解是：系统承担阅读、抽取和关联这些重复劳动，人仍然负责定义策略、评估影响并批准处置。
从 Threat Signal 到 WAF 规则，中间仍需要策略层
自动提取指标并不等于可以自动阻断。生产环境至少还应补充以下判断：
- 可信度门槛：低置信度指标只进入观察队列，高置信度指标才成为规则候选。
- 有效期：攻击基础设施变化很快，指标应设置复查时间或过期时间。
- 资产范围：某条情报是否真的针对当前应用、行业或暴露面。
- 共享基础设施风险：直接封禁云服务、VPN 出口、CDN 节点或 NAT 地址，可能造成大面积误伤。
- 处置分级：先观察命中情况，再根据业务影响升级为质询或阻断。
- 可追溯性：保留来源、事件名称、创建时间、审批人和撤销条件。
因此，Threat Signals 更适合作为 WAF 的“上下文入口”，而不是一个绕过变更流程的自动封禁按钮。
可以这样实践：先生成候选表达式，再人工审核
下面是一个可复制运行的本地示例。它假设已经获得一组结构化指标，并根据置信度生成 Cloudflare WAF 候选表达式。示例中的字段并非官方 Threat Signals 导出格式；接入时应按照实际界面或数据结构调整转换逻辑。
示例使用的 IP 和域名均为文档保留值，不会指向真实攻击目标：
cat > signals.json <<'JSON'
[
{"type": "ip", "value": "198.51.100.42", "confidence": 92, "source": "public-report-a"},
{"type": "cidr", "value": "203.0.113.0/24", "confidence": 88, "source": "public-report-b"},
{"type": "domain", "value": "bad.example", "confidence": 85, "source": "public-report-a"},
{"type": "ip", "value": "not-an-ip", "confidence": 99, "source": "invalid-sample"}
]
JSON
cat > build_waf_candidates.py <<'PY'
import ipaddress
import json
import sys
INPUT = sys.argv[1] if len(sys.argv) > 1 else 'signals.json'
MIN_CONFIDENCE = int(sys.argv[2]) if len(sys.argv) > 2 else 80
def valid_domain(value):
if len(value) > 253:
return False
labels = value.split('.')
return all(
label
and len(label) <= 63
and label[0].isalnum()
and label[-1].isalnum()
and all(ch.isalnum() or ch == '-' for ch in label)
for label in labels
)
with open(INPUT, encoding='utf-8') as handle:
signals = json.load(handle)
ip_values = set()
domains = set()
rejected = []
for signal in signals:
kind = str(signal.get('type', '')).lower()
value = str(signal.get('value', '')).strip().lower()
confidence = int(signal.get('confidence', 0))
if confidence < MIN_CONFIDENCE:
rejected.append((value, 'below confidence threshold'))
continue
try:
if kind == 'ip':
ip_values.add(str(ipaddress.ip_address(value)))
elif kind == 'cidr':
ip_values.add(str(ipaddress.ip_network(value, strict=False)))
elif kind == 'domain' and valid_domain(value):
domains.add(value)
else:
rejected.append((value, 'unsupported or invalid indicator'))
except ValueError:
rejected.append((value, 'invalid IP or CIDR'))
print('# Review these candidates before deployment')
if ip_values:
print('ip.src in {' + ' '.join(sorted(ip_values)) + '}')
if domains:
quoted = ' '.join(json.dumps(item) for item in sorted(domains))
print('http.host in {' + quoted + '}')
if rejected:
print('\n# Rejected indicators')
for value, reason in rejected:
print(f'# {value}: {reason}')
PY
python3 build_waf_candidates.py signals.json 80
预期输出类似：
# Review these candidates before deployment
ip.src in {198.51.100.42 203.0.113.0/24}
http.host in {"bad.example"}
# Rejected indicators
# not-an-ip: invalid IP or CIDR
这段脚本有意只生成候选条件，不直接调用 API 发布规则。实际采用时，可以继续加入允许列表排除、指标过期时间、来源数量、资产标签和审批状态。随后在 WAF 中先选择当前套餐可用的低影响动作观察命中，再决定是否升级处置。
上线时不要省略这些护栏
将 Threat Signals 纳入生产流程时，可以用下面的清单控制风险：
- 为每条规则记录威胁来源、创建时间和复查日期；
- 对办公出口、合作伙伴、监控节点和关键第三方维护允许列表；
- 在阻断前检查历史请求，确认指标是否命中过正常用户；
- 为批量规则设置数量上限，避免一次情报导入扩大故障范围；
- 监控命中率、误报、延迟和业务转化变化；
- 准备快速停用规则的回滚路径；
- 定期清理过期指标，而不是让规则永久累积。
Threat Signals 的价值在于把散落在公开报告中的信息变成可操作、可关联的安全上下文。它能减少复制粘贴和跨团队转交，但无法替代风险判断。最佳落地方式是让自动化负责抽取与整理，让策略门槛、渐进式处置和人工审核负责控制影响面。