| 维度 | 目标 |
|---|---|
| ✅ 语法安全 | 正则合法、无死循环 |
| ✅ 性能安全 | 避免回溯爆炸(ReDoS) |
| ✅ 输入安全 | 限制输入大小、来源 |
| ✅ 权限安全 | 最小权限、隔离执行 |
| ✅ 可观测性 | 日志、异常、执行时间 |
| ✅ 可审计性 | 规则版本化、可回滚 |
在加载 Grok 规则前,必须先编译验证。
Logstash / Grok 通用做法
# 使用 logstash 自带的 grok 校验
bin/logstash -e '
input { stdin {} }
filter {
grok {
match => { "message" => "%{COMBINEDAPACHELOG}" }
}
}
output { stdout {} }
'✅ 若语法错误,启动时即失败
检查是否存在:
(.*)*((a+)?)+❌ 不安全示例:
(a+)+
(.+)*
(.*)+\$✅ 安全改写:
[a-zA-Z0-9]+
[^ ]+验证方法:
import re
re.compile(r'(a+)+b')npm install -g safe-regex
safe-regex '^(a+)+$'❌ 若返回true→ 高危
✅ 返回false→ 相对安全
filter {
grok {
match => { "message" => "%{COMBINEDAPACHELOG}" }
timeout_millis => 50
}
}if [message] =~ /^.{0,4096}$/ {
grok { ... }
}logstash)ps -ef | grep logstash✅ 规则来源:
❌ 禁止:
{ "grok": user_input }直接返回给客户端
✅ 正确做法:
tag_on_failure => ["_grokparsefailure"]mutate {
gsub => [
"password", "[a-zA-Z0-9]+", "***"
]
}关注指标:
_grokparsefailure 数量metrics {
meter => "grok_failures"
}✅ 推荐结构:
grok/
├── patterns/
│ ├── firewall.grok
│ └── web.grok
├── tests/
│ └── web.json
└── CHANGELOG.md✅ 每次变更:
| 风险 | 验证点 |
|---|---|
| ReDoS | 正则无嵌套重复 |
| 无限回溯 | safe-regex 检测 |
| 大日志攻击 | 长度限制 |
| 规则注入 | 静态规则、无用户输入 |
| 权限过高 | 非 root 运行 |
| 信息泄露 | 不暴露异常细节 |
写 Grok
↓
safe-regex 检测
↓
单元测试(异常/超长/边界)
↓
Logstash 本地验证
↓
灰度上线 + 监控
↓
定期审计规则如果你愿意,我可以:
直接把你的 Grok 配置贴出来即可。