Apache日志需通过自定义格式、Filebeat采集、Logstash流式分析实现毫秒级安全告警；核心是将日志转为可触发动作的数据流，含结构化输出、实时捕获、速率统计与Webhook推送。
Apache 日志本身不支持原生实时推送，但可通过轻量级日志采集 + 流式处理链路实现毫秒级安全事件告警。核心在于把日志从“静态记录”变为“可触发动作的数据流”。
一、日志源头：确保结构化与高时效性
避免默认的通用日志格式（如 common），改用含关键安全字段的自定义格式，并启用异步写入：
- 在 /etc/httpd/conf/httpd.conf 中添加：
LogFormat "%h %l %u %t \"%r\" %>s %b \"%{Referer}i\" \"%{User-Agent}i\" %{X-Forwarded-For}i %{UNIQUE_ID}e" secure_combined
CustomLog "|/usr/bin/rotatelogs -f /var/log/httpd/access_realtime.log 5M" secure_combined
- UNIQUE_ID 提供请求唯一标识，便于跨日志关联（如访问日志 + 错误日志 + 应用日志）
- 用 rotatelogs 按大小轮转（而非时间），减少延迟；5M 触发一次切片，兼顾实时性与管理效率
- 确保 httpd 进程对日志目录有写权限：
chown apache:apache /var/log/httpd/
二、采集层：Filebeat 实时捕获并打标
Filebeat 是最轻量、低侵入的实时采集工具，支持事件过滤与字段增强：
Apache Superset 是一个广泛采用的开源 BI 平台，用于 SQL 探索、图表构建和仪表板交付。当代理需要查询仓库数据、组装仪表板或使用成熟的分析界面解释指标而不是临时笔记本代码时，此技能非常有用。
- 配置 /etc/filebeat/filebeat.yml：
filebeat.inputs:
- type: log
enabled: true
paths:
- "/var/log/httpd/access_realtime.log"
fields:
log_type: "apache_access_secure"
processors:
- dissect:
tokenizer: "%{clientip} %{ident} %{auth} [%{timestamp}] \"%{method} %{url} %{protocol}\" %{status} %{bytes} \"%{referer}\" \"%{useragent}\" %{xforwardedfor} %{unique_id}"
- drop_event.when.and:
- contains.field: useragent
- regexp.field: useragent
pattern: "(^$|curl|wget|sqlmap|nikto)"
- dissect 处理器比 grok 更快，适合高吞吐场景；直接提取 clientip、url、status 等字段供后续判断
- drop_event 过滤掉已知扫描工具 User-Agent，降低后端压力
- 输出目标建议直连 Kafka（非 Elasticsearch），为告警逻辑留出缓冲和重试空间
三、实时分析与推送：用 Logstash 或轻量脚本做规则引擎
不推荐在 Kibana 做实时告警（延迟高、资源重），应下沉到流处理层触发动作：
- Logstash 配置示例（/etc/logstash/conf.d/apache_alert.conf）：
filter {
if [log_type] == "apache_access_secure" {
# 检测暴力破解：1分钟内同一IP对/login.*提交失败≥5次
if [method] == "POST" and [url] =~ /\/login.*/ and [status] =~ /^4\d\d$/ {
aggregate {
task_id => "%{clientip}"
code => "map['count'] ||= 0; map['count'] += 1"
timeout => 60
push_map_as_event_on_timeout => true
timeout_tags => ['_alert_bruteforce']
}
}
}
}
output {
if "_alert_bruteforce" in [tags] and [count] >= 5 {
http {
url => "https://your-webhook.example.com/alert"
http_method => "post"
format => "json"
body => '{"event":"brute_force","ip":"%{clientip}","count":%{count},"ts":"%{@timestamp}"}'
}
}
}
- 利用 Logstash aggregate 插件实现窗口计数，无需外部存储即可完成速率统计
- HTTP output 直推企业微信/钉钉机器人或内部告警平台，延迟通常
- 其他常见规则（可复用同框架）：SQL注入关键词（union select）、路径遍历（../）、高频 404 扫描、大体积 POST（>2MB）
四、补充建议：提升可靠性与可观测性
-
防丢日志：Filebeat 启用
backoff和max_retries，Kafka 设置acks=all，确保网络抖动时不丢失事件 - 防误报：对告警事件加二次确认——例如将触发 IP 加入临时黑名单 5 分钟，若期间仍有异常请求再发正式通知
-
可追溯：在告警 payload 中携带
UNIQUE_ID，方便运维人员快速在原始日志中定位完整请求上下文 - 降噪：区分环境——测试/预发环境的扫描行为不告警，仅生产环境启用全部规则
