3. ELK Stack入门：Elasticsearch安装与配置、Logstash管道配置、Kibana可视化基础

说到审计日志分析，ELK Stack 几乎是绕不开的标配。我最早接触它是在一次安全事件应急响应中，客户要求从 200 多台服务器里快速定位异常登录行为。那时候没有 ELK，全靠 grep 和 awk 硬扛，熬了三个通宵。后来换了 ELK，同样的数据量，半小时出结果。嗯，这就是我为什么把这章放在课程靠前位置的原因——工具选对了，事半功倍。

3.1 Elasticsearch：日志的存储与搜索引擎

Elasticsearch 是整个 ELK 的核心。说白了，它就是一个分布式的、基于 JSON 的搜索引擎。审计日志进来之后，全存在这里。你想想看，每天几十 GB 的登录日志、操作日志，如果没有一个高效的存储和检索机制，根本没法玩。

3.1.1 安装与基础配置

我个人习惯用 tar 包安装，这样对目录结构心里有数。当然，生产环境建议用 RPM 或 DEB 包，方便 systemd 管理。这里我演示一下单节点部署，适合咱们实验环境。

# 下载并解压（版本以 7.17.x 为例）
wget https://artifacts.elastic.co/downloads/elasticsearch/elasticsearch-7.17.14-linux-x86_64.tar.gz
tar -xzf elasticsearch-7.17.14-linux-x86_64.tar.gz
cd elasticsearch-7.17.14/

# 修改配置文件
vim config/elasticsearch.yml

关键配置项我列在下面，每个参数我都踩过坑：

配置项	推荐值	说明
cluster.name	audit-log-cluster	集群名称，同一集群必须一致
node.name	node-1	节点名称，建议有业务含义
path.data	/data/elasticsearch	数据目录，别放系统盘
path.logs	/var/log/elasticsearch	日志目录，方便排查问题
network.host	0.0.0.0	监听地址，生产环境要绑定内网 IP
http.port	9200	REST API 端口
discovery.type	single-node	单节点模式，集群环境去掉

3.1.2 启动与验证

ES 不允许用 root 用户启动，这是安全机制。我一般会创建一个专用用户：

# 创建用户并授权
useradd -m -s /bin/bash esuser
chown -R esuser:esuser /opt/elasticsearch-7.17.14/

# 切换到 esuser 启动
su - esuser
cd /opt/elasticsearch-7.17.14/
./bin/elasticsearch -d -p pid

# 验证是否启动成功
curl -X GET "localhost:9200/"

如果返回类似下面的 JSON，说明 ES 跑起来了：

{
  "name" : "node-1",
  "cluster_name" : "audit-log-cluster",
  "cluster_uuid" : "xxxxxx",
  "version" : {
    "number" : "7.17.14",
    "build_flavor" : "default"
  },
  "tagline" : "You Know, for Search"
}

3.2 Logstash：日志的采集与处理管道

Logstash 的作用，就是把各种格式的审计日志「吃进去」，然后「吐出来」变成结构化的 JSON 数据，再送给 ES。我习惯把它想象成一个数据加工厂——原料是乱七八糟的文本，成品是干净整洁的 JSON。

3.2.1 管道配置三要素

每个 Logstash 管道都有三个核心阶段：input、filter、output。我画了一张图，帮你理解这个流程：

3.2.2 实战：解析 SSH 登录日志

这是我在项目中经常用到的场景。假设我们要收集 /var/log/secure 中的 SSH 登录记录，配置如下：

# ssh-login-pipeline.conf
input {
  file {
    path => "/var/log/secure"
    type => "ssh-login"
    start_position => "beginning"
    sincedb_path => "/dev/null"  # 测试环境用，生产环境去掉
  }
}

filter {
  grok {
    match => {
      "message" => "%{SYSLOGTIMESTAMP:timestamp} %{SYSLOGHOST:hostname} sshd\[%{NUMBER:pid}\]: %{GREEDYDATA:event}"
    }
  }
  date {
    match => ["timestamp", "MMM  d HH:mm:ss", "MMM dd HH:mm:ss"]
    target => "@timestamp"
  }
  mutate {
    remove_field => ["message", "timestamp", "hostname"]
  }
}

output {
  elasticsearch {
    hosts => ["localhost:9200"]
    index => "ssh-login-%{+YYYY.MM.dd}"
  }
  stdout { codec => rubydebug }  # 调试用，生产环境去掉
}

3.3 Kibana：数据可视化与探索

Kibana 是 ELK 的「门面」。数据进了 ES，怎么查、怎么看、怎么出报告，全靠它。我个人觉得，Kibana 最大的价值在于让安全运维人员能「看见」日志中的异常模式。

3.3.1 连接 ES 与创建索引模式

安装 Kibana 很简单，解压后改一下 kibana.yml：

server.port: 5601
server.host: "0.0.0.0"
elasticsearch.hosts: ["http://localhost:9200"]
kibana.index: ".kibana"

启动后访问 http://你的IP:5601。第一步是创建索引模式（Index Pattern）。比如我们刚才用 Logstash 写入了 ssh-login-2024.01.01 这样的索引，那么在 Kibana 里就匹配 ssh-login-*。

3.4.2 可视化基础：从数据到图表

Kibana 的可视化类型很丰富，我挑几个审计日志分析最常用的：

• 数据表（Data Table）：展示原始日志，适合逐条排查
• 柱状图（Vertical Bar）：按时间维度统计事件数量，看趋势
• 饼图（Pie）：按事件类型或来源 IP 做占比分析
• 坐标图（Coordinate Map）：如果日志里有经纬度，可以在地图上展示攻击来源

举个例子，我想看过去 24 小时内 SSH 登录失败次数最多的前 10 个 IP：

1. 在 Kibana 里创建一个 Vega-Lite 或 TSVB 可视化
2. 选择索引模式 ssh-login-*
3. 过滤条件：event: "Failed password"
4. 聚合方式：Terms aggregation 对 source_ip 字段计数
5. 排序：按计数降序，取 Top 10

这样一张「暴力破解来源 IP 排行榜」就出来了。我在一次护网行动中，就是靠这个图表在 5 分钟内定位到了攻击者的 C2 服务器。

3.4 本章小结

ELK Stack 的三个组件各司其职：ES 负责存和查，Logstash 负责洗数据，Kibana 负责展示。我建议你在实验环境里亲手搭一遍，把 SSH 日志从采集到可视化完整跑通。这个过程走通了，后面章节的告警规则、自动化响应、溯源分析才有基础。