ELK Stack日志分析入门：搭建安全事件监控平台

作者

适用场景

ELK Stack（Elasticsearch、Logstash、Kibana）是业界最流行的开源日志分析平台，适用于安全运营中心（SOC）建设、服务器日志集中管理、攻击事件溯源分析和合规审计。本教程适用于安全运维人员、渗透测试工程师和系统管理员，帮助快速搭建安全日志分析基础设施。

前置条件

  • 一台Linux服务器（推荐4核8G以上，Ubuntu 20.04+/CentOS 7+）
  • Java运行环境（Elasticsearch 7.x+需要JDK 11+）
  • sudo或root权限
  • 基本的Linux命令行操作经验
  • 服务器已有Nginx/Apache等Web服务在运行（用于产生日志）

原理说明

ELK Stack由三部分组成：Elasticsearch（分布式搜索引擎，负责日志存储和检索）、Logstash（数据处理管道，负责日志采集、解析和转发）和Kibana（可视化平台，负责数据展示和告警）。配合Filebeat（轻量级日志采集器）形成完整的ELK + Filebeat架构。安全分析场景中，ELK可集中收集Nginx访问日志、系统认证日志（/var/log/auth.log）、防火墙日志等，通过Kibana仪表盘实时监控攻击行为。

操作步骤

步骤1：安装Elasticsearch

# 导入Elasticsearch GPG密钥
wget -qO - https://artifacts.elastic.co/GPG-KEY-elasticsearch | sudo gpg --dearmor -o /usr/share/keyrings/elasticsearch-keyring.gpg

# 添加APT源
echo "deb [signed-by=/usr/share/keyrings/elasticsearch-keyring.gpg] https://artifacts.elastic.co/packages/8.x/apt stable main" | sudo tee /etc/apt/sources.list.d/elastic-8.x.list

# 安装
sudo apt update && sudo apt install -y elasticsearch

# 配置Elasticsearch（安全日志分析推荐配置）
sudo tee -a /etc/elasticsearch/elasticsearch.yml << 'EOF'
cluster.name: security-log-cluster
node.name: node-1
path.data: /var/lib/elasticsearch
path.logs: /var/log/elasticsearch
network.host: 127.0.0.1
http.port: 9200
xpack.security.enabled: false
discovery.type: single-node
EOF

# 启动服务
sudo systemctl enable elasticsearch
sudo systemctl start elasticsearch
# 验证
curl http://127.0.0.1:9200

步骤2：���装Kibana

sudo apt install -y kibana
sudo tee -a /etc/kibana/kibana.yml << 'EOF'
server.port: 5601
server.host: "127.0.0.1"
elasticsearch.hosts: ["http://127.0.0.1:9200"]
EOF

sudo systemctl enable kibana
sudo systemctl start kibana
# 验证：curl http://127.0.0.1:5601/api/status

步骤3：安装Logstash

sudo apt install -y logstash

# 创建Nginx日志解析配置
sudo tee /etc/logstash/conf.d/nginx-security.conf << 'EOF'
input {
  beats {
    port => 5044
  }
}

filter {
  if [fileset][module] == "nginx" {
    grok {
      match => { "message" => "%{IPORHOST:client_ip} - %{DATA:user_name} [%{HTTPDATE:timestamp}] \"%{WORD:method} %{DATA:request} HTTP/%{NUMBER:http_version}\" %{NUMBER:response_code} %{NUMBER:bytes} \"%{DATA:referrer}\" \"%{DATA:user_agent}\"" }
    }
    date {
      match => [ "timestamp", "dd/MMM/yyyy:HH:mm:ss Z" ]
    }
    # 提取攻击特征字段
    geoip {
      source => "client_ip"
      target => "geo"
    }
    # 标记可疑请求
    if [response_code] == 403 or [response_code] == 401 {
      mutate { add_tag => [ "blocked_request" ] }
    }
    if [request] =~ /(?:select|union|exec|insert|script|alert|eval)/i {
      mutate { add_tag => [ "attack_attempt" ] }
    }
  }
}

output {
  elasticsearch {
    hosts => ["127.0.0.1:9200"]
    index => "nginx-logs-%{+YYYY.MM.dd}"
  }
}
EOF

sudo systemctl enable logstash
sudo systemctl start logstash

步骤4：安装Filebeat采集Nginx日志

sudo apt install -y filebeat

# 配置Filebeat
sudo tee /etc/filebeat/filebeat.yml << 'EOF'
filebeat.inputs:
- type: log
  enabled: true
  paths:
    - /var/log/nginx/access.log
  fields:
    log_type: nginx_access
  # 多行匹配：将同一请求的日志合并
  multiline.pattern: '^\d{1,3}\.\d{1,3}\.'
  multiline.negate: true
  multiline.match: after

output.logstash:
  hosts: ["127.0.0.1:5044"]
EOF

sudo systemctl enable filebeat
sudo systemctl start filebeat

步骤5：采集系统认证日志

# 添加auth.log采集配置
sudo tee -a /etc/logstash/conf.d/auth-security.conf << 'EOF'
input {
  file {
    path => "/var/log/auth.log"
    type => "auth"
    start_position => "beginning"
  }
}

filter {
  if [type] == "auth" {
    grok {
      match => { "message" => "%{SYSLOG5424PRI:priority}?%{SYSLOGLINE}" }
    }
    # 标记SSH暴力破解尝试
    if "Failed password" in [message] {
      mutate { add_tag => [ "ssh_bruteforce" ] }
    }
    if "Accepted password" in [message] {
      mutate { add_tag => [ "ssh_success" ] }
    }
  }
}

output {
  elasticsearch {
    hosts => ["127.0.0.1:9200"]
    index => "auth-logs-%{+YYYY.MM.dd}"
  }
}
EOF

sudo systemctl restart logstash

步骤6：在Kibana中创建安全仪表盘

# 通过API创建索引模式（用于Kibana）
curl -X POST "http://127.0.0.1:5601/api/saved_objects/index-pattern" \
  -H "Content-Type: application/json" \
  -H "kbn-xsrf: true" \
  -d '{
    "attributes": {
      "title": "nginx-logs-*",
      "timeFieldName": "@timestamp"
    }
  }'

curl -X POST "http://127.0.0.1:5601/api/saved_objects/index-pattern" \
  -H "Content-Type: application/json" \
  -H "kbn-xsrf: true" \
  -d '{
    "attributes": {
      "title": "auth-logs-*",
      "timeFieldName": "@timestamp"
    }
  }'

配置验证

# 验证Elasticsearch索引
curl -X GET "http://127.0.0.1:9200/_cat/indices?v"

# 验证日志是否写入
curl -X GET "http://127.0.0.1:9200/nginx-logs-$(date +%Y.%m.%d)/_search?pretty" -d '{"size": 5}' -H "Content-Type: application/json"

# 查看攻击标记（401/403请求数）
curl -X GET "http://127.0.0.1:9200/nginx-logs-*/_count" -d '{
  "query": { "term": { "tags": "blocked_request" } }
}' -H "Content-Type: application/json"

常见问题

Q1：Filebeat无法连接到Logstash（Connection refused）怎么办？

检查三个因素：1）Logstash是否正在运行：sudo systemctl status logstash；2）Logstash监听端口：sudo netstat -tlnp | grep 5044；3）防火墙是否开放端口：sudo ufw status。如果Logstash未绑定0.0.0.0，需在input配置中添加host => "0.0.0.0"参数。

Q2：大量日志导致Elasticsearch磁盘空间不足如何处理？

实施索引生命周期管理（ILM）：在Elasticsearch中配置ILM策略，自动将7天前的索引转为只读，30天前的索引删除。或者使用cron定时任务：0 3 * * * curl -X DELETE "http://127.0.0.1:9200/nginx-logs-$(date -d '30 days ago' +%Y.%m.%d)"。同时建议在Filebeat层对高流量站点做日志采样（sampling: 0.1）。

总结

ELK Stack为安全日志分析提供了强大的基础设施，通过集中采集Nginx访问日志和系统认证日志，配合Grok解析、GeoIP地理定位和自动攻击标记，安全团队可以实时发现攻击行为、追溯攻击来源。建议在此基础上进一步配置Kibana Alert告警规则，对SSH暴力破解（5分钟内超过10次失败登录）和SQL注入尝试发送邮件或钉钉通知，实现主动安全防御。