ElastAlert监控日志告警Web攻击行为 - 详细教学文档

一、ELK环境搭建

1.1 ELK组件介绍

• Elasticsearch：分布式、可扩展的实时搜索与数据分析引擎
• Logstash：轻量级日志搜集处理框架，可自定义处理日志并传输
• Kibana：开源分析与可视化平台，用于展示Elasticsearch数据

三者关系：

• Logstash（Controller层）：数据收集和过滤
• Elasticsearch（Model层）：数据存储和索引
• Kibana（View层）：数据可视化和展示

1.2 版本选择与安装

重要版本限制：

• ElastAlert不支持Elasticsearch 6.0+版本
• 推荐版本组合：
  • Elasticsearch: 5.5.2
  • Kibana: 5.5.2
  • Logstash: 6.0.0
  • Filebeat: 6.0.0

安装方法：

• Ubuntu系统使用deb包安装：

  dpkg -i package.deb
  

• CentOS系统使用rpm包安装

1.3 ELK配置

Elasticsearch配置

/etc/elasticsearch/elasticsearch.yml:

network.host: 127.0.0.1
http.port: 9200

Kibana配置

/etc/kibana/kibana.yml:

server.port: 5601
server.host: "localhost"
elasticsearch.url: "http://localhost:9200"

Logstash配置

示例配置文件filebeat_log.conf:

input {
  beats {
    port => 5044
    client_inactivity_timeout => 90
    codec => json
  }
}

filter {
  date {
    match => ["timestamp", "dd/MMM/yyyy:HH:mm:ss Z"]
    target => ["datetime"]
  }
  geoip {
    source => "remote_addr"
  }
  mutate {
    remove_field => ["tags", "beat"]
  }
}

output {
  elasticsearch {
    hosts => "localhost:9200"
    index => "logstash-%{+YYYY.MM.dd}"
  }
  stdout {}
}

1.4 ELK启动

启动Elasticsearch和Kibana：

systemctl daemon-reload
systemctl enable elasticsearch.service
systemctl enable kibana.service
systemctl start elasticsearch
systemctl start kibana

启动Logstash：

nohup ./logstash -f filebeat_log.conf > /dev/null 2>&1 &

二、日志收集架构

2.1 架构设计

[被监控服务器A/B/C...] → [主服务器(ELK+ElastAlert)]

使用Filebeat作为轻量级日志收集器，Logstash作为中心过滤器

2.2 Filebeat配置

示例filebeat.yml：

filebeat.prospectors:
- type: log
  enabled: true
  paths:
    - /path/to/logs/*.txt
  document_type: tomcat-log
  scan_frequency: 15s
  ignore_older: 20m
  close_inactive: 12m
  clean_inactive: 30m
  close_removed: true
  clean_removed: true

output.logstash:
  hosts: ["logstash_server:5044"]

启动Filebeat：

nohup ./filebeat -e -c filebeat.yml >/dev/null 2>&1 &

三、日志格式转换

3.1 Tomcat JSON格式配置

修改server.xml：

<Valve className="org.apache.catalina.valves.AccessLogValve"
  directory="logs"
  prefix="localhost_access_log"
  suffix=".txt"
  pattern='{"time":"%t","remote_addr":"%h","remote_user":"%l","request":"%r","status":"%s","body_bytes_sent":"%b","http_referer":"%{Referer}i","http_user_agent":"%{User-Agent}i","http_x_forwarded_for":"%{X-Forwarded-For}i","request_time":"%T","host":"%v","port":"%p"}' />

3.2 Nginx JSON格式配置

在nginx.conf中添加：

log_format logstash_json '{ "time" : "$time_local", '
  '"remote_addr" : "$remote_addr", '
  '"remote_user" : "$remote_user", '
  '"request" : "$request", '
  '"status" : "$status", '
  '"body_bytes_sent" : "$body_bytes_sent", '
  '"http_referer" : "$http_referer", '
  '"http_user_agent" : "$http_user_agent", '
  '"http_x_forwarded_for" : "$http_x_forwarded_for", '
  '"request_time" : "$request_time", '
  '"request_length" : "$request_length", '
  '"host" : "$http_host" }';

四、ElastAlert配置与使用

4.1 安装ElastAlert

git clone https://github.com/Yelp/elastalert.git
cd elastalert
python setup.py install
pip install -r requirements.txt
cp config.yaml.example config.yaml

4.2 创建索引

elastalert-create-index

4.3 主配置文件config.yaml

rules_folder: example_rules
run_every:
  seconds: 3
buffer_time:
  minutes: 15
es_host: 188.88.88.88
es_port: 9200
writeback_index: elastalert_status
alert_time_limit:
  days: 1

4.4 告警规则配置

示例webattack_frequency.yaml：

name: web attack
realert:
  minutes: 5
type: frequency
index: logstash-*
num_events: 10
timeframe:
  minutes: 1

filter:
- query_string:
    query: "request: select.+(from|limit) OR request: union(.*?)select OR request: into.+(dump|out)file OR request: (base64_decode|sleep|benchmark|and.+1=1|and.+1=2|or %20|exec|information_schema|where %20|union %20|%2ctable_name %20|cmdshell|table_schema) OR request: (iframe|script|body|img|layer|div|meta|style|base|object|input|onmouseover|onerror|onload) OR request: .+etc.+passwd OR http_user_agent:(HTTrack|harvest|audit|dirbuster|pangolin|nmap|sqln|-scan|hydra|Parser|libwww|BBBike|sqlmap|w3af|owasp|Nikto|fimap|havij|PycURL|zmeu|BabyKrokodil|netsparker|httperf|bench) OR status: (400|404|500|501) NOT (request:_health.html OR remote_addr:222.222.222.222)"

smtp_host: smtp.qiye.163.com
smtp_port: 25
smtp_auth_file: /path/to/smtp_auth_file.yaml
email_reply_to: xxx@163.com
from_addr: xxx@163.com
alert:
- "email"
email:
- "recipient@example.com"
alert_subject: "web attack may be by {} at @{}"
alert_subject_args:
- remote_addr
- time
alert_text_type: alert_text_only
alert_text: |
  你好,服务器({})可能正在受到web攻击,请采取手段阻止!!!!
  ### 截止发邮件前匹配到的请求数:{}
  > 发生时间: {}
  > timestamp:{}
  > attacker's ip: {}
  > request: {}
  > status:{}
  > UA头:{}
  >>> 参考来源:{}  
alert_text_args:
- host
- num_hits
- time
- "@timestamp"
- remote_addr
- request
- status
- http_user_agent
- source

SMTP认证文件smtp_auth_file.yaml：

user: xxx@163.com
password: yourpassword

4.5 启动ElastAlert

nohup python -m elastalert.elastalert --verbose --rule webattack_frequency.yaml >/dev/null 2>&1 &

五、告警效果示例

收到的告警邮件示例：

主题: web attack may be by 104.38.13.21 at @[13/Jan/2018:16:06:58 +0800]

内容:
你好,服务器(199.222.36.31)可能正在受到web攻击,请采取手段阻止!!!!
### 截止发邮件前匹配到的请求数:20
> 发生时间: [13/Jan/2018:16:06:58 +0800]
> timestamp: 2018-01-13T08:07:04.930Z
> attacker's ip: 184.233.9.121
> request: GET /dbadmin/scripts/setup.php HTTP/1.0
> status: 200
> UA头: ZmEu
>>> 参考来源: /log/localhost_access_log.2018-01-13.txt

六、最佳实践与注意事项

1. 版本兼容性：确保ElastAlert与Elasticsearch版本兼容
2. 性能考虑：使用Filebeat代替直接使用Logstash收集日志
3. 日志格式：统一使用JSON格式便于分析
4. 规则优化：根据实际攻击模式调整检测规则
5. 告警频率：合理设置realert时间避免告警风暴
6. 测试验证：使用elastalert-test-rule测试规则有效性
7. 错误处理：遇到索引问题时可以删除并重建索引

七、扩展功能

1. 其他告警方式：可集成微信、钉钉等告警渠道
2. 其他规则类型：ElastAlert支持11种告警规则，可根据需求选择
3. 可视化增强：结合Kibana Dashboard进行攻击可视化分析
4. 守护进程：使用Supervisor管理ElastAlert进程

通过以上配置，可以构建一个完整的Web攻击监控和告警系统，及时发现并响应潜在的安全威胁。