多彩编程 多彩编程MZPH · CODE BLOG
ARTICLE DETAIL

文章详情

深耕前端与后端开发技术的一线实战笔记与踩坑复盘。

ELK安全可观测实战:从日志采集到告警的完整闭环

ELK安全可观测实战:从日志采集到告警的完整闭环 先把话说在前面这个系列不是教你怎么把 ELK 装起来看一眼日志就完事而是把它当成安全建设里的一层“防护记录仪”来用。很多人觉得 ELK 就是日志聚合、画几个 Kibana 图表真正遇到安全事件时才发现日志链路是断的、字段是乱的、告警是滞后的。我写这个开篇就是想把“日志”从被动留痕升级成主动防御的一部分让 ELK 真正参与到安全可观测体系里而不是只在排查问题时才被想起来。这个系列适合谁看三类人一是刚接手安全运维、需要把分散日志统一收拢的工程师二是已经在用 ELK 做业务日志、想往安全方向延伸的开发者三是团队里负责安全合规、需要给审计和应急响应提供数据依据的负责人。你不需要一开始就懂全部原理我会从最核心的链路讲起把每个环节为什么这么做、踩过哪些坑都说清楚。1. 为什么说日志是“防弹衣”的一部分1.1 安全建设里最容易被忽视的“最后一公里”安全防护通常分成几层边界防火墙拦截、主机加固、应用层 WAF、身份认证、数据加密。这些都属于“主动防御”目的是在攻击发生前或攻击进行中把人挡住。但任何防护都有绕过可能零日漏洞、内部人员误操作、配置错误、供应链攻击这些不一定能被主动防御完全识别。真正决定一个安全事件是“小事故”还是“大灾难”的往往是事后能不能快速定位、还原、取证。而这一切的基础就是日志。我见过很多团队安全设备买了一大堆告警平台也接了但真到应急响应时发现三件事第一关键主机没接入日志采集第二日志字段不统一不同来源的时间格式、IP 字段、用户名字段对不上第三日志保留周期太短攻击者早就把痕迹清理了你连入侵路径都还原不出来。这些问题的本质不是工具不够好而是日志没有被当作安全基础设施来设计。1.2 可观测性与安全的交汇点可观测性Observability这个概念最早更多用于应用性能监控包含 Metrics、Logging、Tracing 三支柱。安全领域以前习惯叫“日志审计”“安全信息与事件管理SIEM”但随着系统规模变大、攻击手法变复杂安全和可观测性正在快速融合。原因很简单安全和业务跑在同一套基础设施上安全数据本质上是业务运行数据的一部分。与其单独搭一套安全日志平台不如基于统一的日志底座在采集、清洗、存储、检索、告警的每个环节叠加安全视角。举个例子一次暴力破解攻击从可观测性视角看是“登录失败次数异常升高”“某个账号短时间内多次认证失败”“源 IP 访问频率异常”从安全视角看就是“正在发生的攻击行为”。两套体系处理的是同一批日志只是分析角度不同。用 ELK 做安全可观测就是先把这些日志统一收进来再用安全规则去筛、去关联、去告警。1.3 ELK 在安全场景里的定位与边界ELK 是 Elasticsearch、Logstash、Kibana 的合称后来 Beats 系列采集器也并入了这个生态所以很多人会说“Elastic Stack”。它在安全场景里通常被用来做日志集中管理、检索分析、告警通知也有人拿它做轻量级 SIEM。但要说清楚ELK 不是万能的。它不是终端检测与响应EDR产品不能直接拦截攻击它不是完整的 SIEM缺乏很多商业 SIEM 内置的合规报表和威胁情报联动。它的价值在于给你一个完全可控、灵活定制、成本相对可控的日志分析底座。你可以自己定义索引结构、自己写告警规则、自己接任何数据源。这种灵活性在企业安全建设里非常重要因为安全日志来源太杂了商业 SIEM 不一定能覆盖所有定制需求而 ELK 可以。2. 核心链路拆解从采集到告警的完整闭环2.1 数据采集层Filebeat 为主Logstash 为辅安全日志的采集是整个链路的地基。采集层没做好后面分析、告警全白搭。最常用的采集器是 Filebeat它轻量、占用资源小、支持多种输入源比如文件、标准输出、TCP/UDP、云平台审计日志等。部署方式一般是在每台需要采集的主机上装一个 Filebeat Agent把指定路径下的日志文件读取后发送到 Logstash 或直接发到 Elasticsearch。这里有个重要选型逻辑为什么不直接用 Logstash 采集因为 Logstash 是 JVM 应用内存占用高如果每台主机都装一个 Logstash资源开销太大。Filebeat 使用 Go 编写内存占用通常在几十 MB 级别适合作为轻量级 Agent 分发到大量主机。Logstash 更适合部署在服务端集中做数据解析、清洗、富化。用生活化类比Filebeat 是各个门店的前台只负责把客人信息登记好送上来Logstash 是总部数据中台负责把各种格式的登记表统一成标准格式。两者分工不同配合使用才能兼顾性能与灵活性。2.2 解析清洗层Logstash 的 Grok 与 Dissect日志到了 Logstash 之后第一件事是解析。不同来源的日志格式千差万别系统日志可能是“时间 主机 进程: 消息”的格式Web 访问日志可能是 Apache/Nginx 的 combined 格式安全设备日志可能是 keyvalue 格式应用日志可能是 JSON 格式。Logstash 最强大的地方就是可以用 Grok 正则模板把这些非结构化字符串拆成结构化字段。Grok 本质上是把正则表达式封装成可复用的模式比如%{IP:client_ip}可以匹配 IP 地址并赋值给client_ip字段%{TIMESTAMP_ISO8601:log_timestamp}可以匹配标准时间戳。但 Grok 有个性能问题正则匹配是 CPU 密集型操作如果日志量大、规则复杂很容易成为瓶颈。所以我建议能不用 Grok 就不用 Grok优先用 Dissect。Dissect 是固定分隔符解析类似于按位置切分字符串性能比 Grok 高一个数量级以上。举个例子如果日志格式固定是[2025-01-01 12:00:00] [INFO] [User:12345] Login success用 Dissect 写[%{ts}] [%{level}] [User:%{user_id}] %{message}就能直接解析完全不需要正则。但要注意Dissect 的前提是格式高度固定。如果同一类日志里某些字段会出现也可能不出现Dissect 会解析失败这时候就得用 Grok 处理可选部分。实际项目中我通常的做法是先画一遍日志样本看格式固定程度。能固定就用 Dissect不能固定就用 Grok再不行就 Grok 和 Dissect 混合用。2.3 存储检索层Elasticsearch 索引设计与生命周期Elasticsearch 是整个 ELK 的性能核心。很多人觉得 ES 慢其实大多数情况下是索引设计不合理。安全日志有个特点写入量大、保留周期长、查询模式相对固定。如果按照默认的按天索引来存时间久了会产生大量小索引集群分片数量膨胀性能急剧下降。我推荐的安全日志索引设计策略是按数据源分大类再按时间分索引。例如sec-log-auth-2025.01.01、sec-log-web-2025.01.01、sec-log-firewall-2025.01.01。这样每个索引的数据量相对可控查询时可以指定索引模式避免全集群扫描。索引生命周期管理ILM一定要用起来。ILM 可以自动完成索引从热阶段到温阶段再到删除阶段的流转。比如热阶段保存最近 3 天数据使用 SSD 高性能节点温阶段保存最近 30 天数据使用 HDD 大容量节点30 天之后自动删除或者归档到冷存储。这个策略能有效控制存储成本同时保证近期数据查询速度。如果不做 ILM等集群磁盘爆了再手动删索引就晚了。2.4 可视化与告警Kibana 的 Lens 与 AlertingKibana 在安全场景里的作用不止是画图。它有几个核心能力第一是 Discover 模块做日志检索和字段探索第二是 Lens 做可视化仪表板第三是 Alerting 模块做告警规则配置第四是 Security 模块如果接了 Endpoint Security 数据做安全事件关联分析。对于安全日志分析我建议不要一上来就追求复杂仪表板。先做几个最实用的视图登录失败趋势、各源 IP 访问量 Top N、Web 攻击特征命中次数、防火墙拒绝事件趋势、账号锁定事件列表。这些视图能覆盖 80% 的日常安全监控需求。告警方面Kibana Alerting 支持基于查询条件触发告警比如“15 分钟内登录失败次数超过 50 次”就触发告警。告警通知可以接入钉钉、企业微信、Slack、邮件等渠道。这里有个经验告警规则不要一开始就设得很灵敏。阈值设太低会刷屏安全运维人员很快会麻木真正重要的告警也会被淹没。我见过一个团队把“登录失败 3 次”就设成告警结果一天几千条告警根本没人看。合理的做法是先设一个较宽的阈值跑两周观察正常基线再根据基线调整。3. 实操过程搭建一套最小可用安全日志平台3.1 环境规划与组件选型先说一个常见误区很多人一上来就要装全套 Elastic Stack其实可以先从最小可用架构开始。我推荐的起步方案是1 台服务器部署 Elasticsearch、Logstash、Kibana另外在需要采集日志的主机上安装 Filebeat。后期数据量大了再拆分节点。版本选择上尽量使用同一大版本的组件。比如全部用 8.x或者全部用 7.17.x。不同版本之间的兼容性差异会导致很多莫名其妙的问题。我在实际项目中碰到过 Filebeat 7.x 往 Elasticsearch 8.x 发数据结果报 authentication 相关错误折腾了半天才发现是版本兼容问题。所以再次强调版本统一很重要。部署方式有三种可选原生安装包、Docker Compose、HelmKubernetes 环境。如果是测试环境Docker Compose 最方便如果是生产环境我更推荐原生安装包或者容器化平台统一管理。Docker 跑 ELK 的问题是容器重启后数据持久化容易搞错需要挂载卷路径配好另外 Elasticsearch 对内存和文件句柄的要求在容器里一不留神就会踩坑。3.2 用 Docker Compose 快速拉起 ELK 核心组件这里给一个可以直接复制的参考配置适合 8.x 版本。先建一个docker-compose.yml文件内容如下version: 3.8 services: elasticsearch: image: docker.elastic.co/elasticsearch/elasticsearch:8.10.2 container_name: elasticsearch environment: - discovery.typesingle-node - ES_JAVA_OPTS-Xms1g -Xmx1g - xpack.security.enabledfalse ports: - 9200:9200 volumes: - es_data:/usr/share/elasticsearch/data networks: - elk logstash: image: docker.elastic.co/logstash/logstash:8.10.2 container_name: logstash ports: - 5044:5044 volumes: - ./logstash/pipeline:/usr/share/logstash/pipeline depends_on: - elasticsearch networks: - elk kibana: image: docker.elastic.co/kibana/kibana:8.10.2 container_name: kibana environment: - ELASTICSEARCH_HOSTShttp://elasticsearch:9200 ports: - 5601:5601 depends_on: - elasticsearch networks: - elk volumes: es_data: networks: elk: driver: bridge注意这里把xpack.security.enabled设成了false是为了测试环境方便。生产环境一定要开启安全认证否则集群直接暴露在网络上相当于把日志裸奔给别人看反而成了安全隐患。这个问题我在后面“常见问题”里专门讲。初始化命令很简单docker compose up -d启动后访问http://服务器IP:5601就能打开 Kibana。Elasticsearch 在http://服务器IP:9200。如果你用云服务器记得在安全组里把这两个端口放行但生产环境不建议把 9200 端口直接暴露公网。3.3 Filebeat 采集系统认证日志并输出到 Logstash假设你要采集 Linux 主机的/var/log/secureCentOS/RHEL 系统或/var/log/auth.logUbuntu/Debian 系统这是记录 SSH 登录、sudo 提权、用户切换等安全事件的核心日志文件。在需要采集的主机上安装 Filebeat以 CentOS 为例rpm -ivh filebeat-8.10.2-x86_64.rpm然后修改/etc/filebeat/filebeat.yml核心配置如下filebeat.inputs: - type: filestream id: auth-log enabled: true paths: - /var/log/secure parsers: - ndjson: keys_under_root: true output.logstash: hosts: [你的LogstashIP:5044]这里用的是filestream类型输入而不是旧版的log类型输入。filestream是 Filebeat 7.13 之后引入的新输入方式维护每个文件的读取状态更可靠支持字段id来标识数据流。如果你用旧版log类型也能工作但建议新项目统一用filestream。配置完成后启动systemctl start filebeat systemctl enable filebeat看到这里你可能有个疑问为什么 Filebeat 不直接把数据发到 Elasticsearch而是要先发到 Logstash因为系统日志原始格式是文本行需要在 Logstash 里做解析清洗把“一段字符串”变成“多个有名字的字段”。如果直接发到 ES后面检索和可视化都会很痛苦。所以输出到 Logstash 是常规做法。3.4 Logstash 解析/var/log/secure并写入 Elasticsearch在 Logstash 所在服务器的./logstash/pipeline/目录下创建配置文件security.conf内容如下input { beats { port 5044 } } filter { if [log][file][path] /var/log/secure or [log][file][path] /var/log/auth.log { grok { match { message %{SYSLOGTIMESTAMP:timestamp} %{SYSLOGHOST:hostname} %{DATA:process}\[%{POSINT:pid}\]: %{GREEDYDATA:message_content} } } date { match [ timestamp, MMM dd HH:mm:ss, MMM d HH:mm:ss ] target timestamp } mutate { rename { message_content message } remove_field [ timestamp, process, pid ] } } } output { elasticsearch { hosts [http://elasticsearch:9200] index sec-log-auth-%{YYYY.MM.dd} } }这段配置做的事情是接收 Filebeat 传来的数据如果日志来源是/var/log/secure或/var/log/auth.log就用 Grok 把系统日志格式拆开提取出时间、主机名、进程名、进程 ID 和实际消息内容然后用date插件把日志里的原始时间转换成标准timestamp最后清理掉多余的字段写入按天切分的sec-log-auth-索引。关于 Grok 匹配失败的问题我在实践中经常遇到。系统日志的时间戳格式在月份和日期之间可能有多个空格比如Jan 1和Jan 12正则如果写死一个空格就会匹配失败。所以date里我写了两种格式MMM dd HH:mm:ss和MMM d HH:mm:ss分别对应单数日期和双数日期。这类细节不实测很难发现但你跑一周日志就能碰到。3.5 Kibana 创建索引模式并建立第一个安全仪表板启动 Kibana 后第一步是进入 Management Stack Management Index Patterns创建sec-log-auth-*索引模式。创建时选择timestamp作为时间字段。然后在 Discover 页面选择这个索引模式就能看到已经解析好的认证日志了。接着建议建立一个叫“认证安全监控”的仪表板。用 Lens 创建如下可视化登录失败次数趋势柱状图X 轴用timestamp按 1 小时分桶Y 轴统计event.outcome: failure的文档数。失败登录来源 IP Top 10表格或条形图按source.ip分组统计。SSH 登录成功趋势折线图按timestamp统计message: Accepted的文档数。用户维度登录分布饼图按user.name分组统计。画这些图表不用写代码Lens 拖拽就能完成。但前提是字段已经解析好。所以前面说的 Grok 解析和索引模式创建是基础仪表板只是展示层。3.6 配置第一条安全告警规则Kibana 8.x 的 Alerting 在 Stack Management Alerts and Insights Rules 里。创建一个 Elasticsearch query 规则的步骤点击 Create rule选择 Elasticsearch query。定义查询条件。例如检测暴力破解输入 KQLmessage: Failed password。设置时间范围最近 15 分钟。设置触发条件count 20。选择通知渠道比如配置 Webhook 到企业微信机器人。规则名称建议带业务语义比如“SSH 暴力破解检测 - 生产主机”。如果后续要调整阈值直接改规则就行不用改代码。告警通知里最好带上查询链接方便收到告警的人一键跳转 Kibana 查看原始日志。这里有个容易忽略的点告警规则的时间范围要跟查询语句配合好否则会出现告警和实际日志对不上的情况。比如你写“最近 15 分钟 Failed password 超过 20 次”那查询里的时间字段必须是timestamp且范围是 now-15m 到 now不要遗漏时间过滤条件。4. 常见问题与排查技巧实录4.1 Filebeat 有数据但 Kibana 没有这是新手最容易遇到的问题。首先确认 Filebeat 进程是否在跑systemctl status filebeat然后看 Filebeat 是否成功连接 Logstashjournalctl -u filebeat -f常见的报错有connection refused连接被拒和EOF对端关闭连接。前者一般是网络不通或端口没开后者通常是 Logstash 没起来或配置有误。如果 Filebeat 日志显示发送成功但 ES 里还是没有数据检查 Logstash 日志docker logs logstash -f常见原因是 Grok 匹配失败日志被丢弃或进入_grokparsefailure标签。解决办法是在 Kibana 的 Discover 里搜索tags: _grokparsefailure把没解析成功的原始日志捞出来对着日志格式调整 Grok 表达式。4.2 时间字段差 8 小时这是 ELK 新手绕不过去的坑。Kibana 默认显示 UTC 时间如果你在中国看到的时间会比本地时间少 8 小时。解决办法有两种一是调整浏览器偏好设置在 Kibana 的 Stack Management Advanced Settings 里把dateFormat:tz改成Asia/Shanghai二是在 Logstash 的date插件中设置timezone Asia/Shanghai。这两步都做的话更稳妥确保日志解析时按指定时区处理原始时间Kibana 展示时也按本地时区显示。需要特别说明的是Elasticsearch 内部存储的timestamp永远是 UTC 时间这是规范不建议改。展示层的时区调整不会影响存储值所以放心改 Kibana 设置即可。4.3 生产环境没有开启安全认证这个问题必须单独强调。很多教程为了图省事把xpack.security.enabledfalse写进配置让新手直接跳过认证。但在生产环境这等于把日志数据赤裸裸地暴露在网络上。Elasticsearch 默认端口 9200如果云服务器安全组配置不当攻击者可以直接访问你的 ES 接口把所有日志数据拉走甚至能删除索引。我见过真实的案例某公司的 ES 因为未开启认证被勒索者发现后用脚本清空了所有索引还留下要赎金的说明。所以生产环境一定要开启安全认证至少做到以下几点启用xpack.security.enabledtrue为内置用户设置强密码通过 TLS 加密传输关闭公网直接访问 ES 端口只让内网或代理访问。4.4 磁盘空间暴涨怎么控制日志平台最大的运维压力就是磁盘。Elasticsearch 默认的副本数是 1意味着每份数据都存两份。如果数据量很大副本会显著增加存储占用。在测试阶段可以先设index.number_of_replicas: 0以减少开销生产环境至少要保留 1 个副本防止节点故障丢数据。更关键的是 ILM 策略。在 Kibana 的 Stack Management Index Lifecycle Policies 里创建一个策略例如热阶段 2 天、温阶段 15 天、删除阶段 30 天。然后把这个策略应用到sec-log-auth-*索引模板上。这样新创建的索引都会自动套用不用每次手动清理。如果已经积压了大量历史数据可以先手动删除旧索引curl -X DELETE http://localhost:9200/sec-log-auth-2024.01.01但要记住删了就没有了删除前务必确认不再需要这些日志或者已经备份。4.5 索引分片过多导致集群变慢这也是常见问题。默认情况下 ES 每个索引 1 个分片但如果日志量很大且按天分索引长时间运行后分片数量会非常多集群性能会明显下降。我见过一个集群分片数量超过几千每个分片都有独立的内存和线程开销查询延迟肉眼可见地上升。解决思路有两种。一种是调整索引模板把number_of_shards设成 3 或 5让单个索引分散到多个节点并行处理。另一种是减少索引数量比如日志量小的话可以按周分索引而不是按天分索引。分片数量的经验公式是分片数 节点数 × 1~3不要盲目追求多分片。5. 进阶方向从日志平台迈向安全可观测5.1 接入更多数据源防火墙、WAF、主机审计基础认证日志搭好之后下一步是扩大数据源覆盖。常见的安全数据源包括防火墙日志来源 IP、目的 IP、端口、动作、策略名称。WAF 日志请求 URL、攻击类型、拦截动作、客户端 IP。主机审计日志进程启动、文件变更、账号变更、计划任务变更。数据库审计日志SQL 语句、登录账号、执行结果。容器安全日志镜像扫描结果、运行时异常行为。每个数据源都有不同的日志格式和字段语义需要分别定制 Logstash 解析规则。建议在索引命名上就区分好比如sec-log-firewall-*、sec-log-waf-*、sec-log-audit-*这样检索和权限管理都方便。5.2 用 Elastic Security 做关联分析Elastic 官方提供了 Security 解决方案Elastic Security它可以基于已经采集到 ES 的日志做检测规则匹配。内置了很多现成的检测规则比如“多次失败登录后成功登录”“可疑的 PowerShell 执行”“创建了 SSH 密钥但未添加注释”等。你可以把 Filebeat 采集的主机日志接入 Elastic Security它会自动建立主机时间线把相关事件串联起来。使用 Elastic Security 时需要注意它依赖 ECSElastic Common Schema字段规范。如果日志字段没有映射到 ECS检测规则可能匹配不到。所以在 Logstash 解析阶段尽量将字段名对齐到 ECS比如源 IP 用source.ip目标 IP 用destination.ip用户用user.name事件结果用event.outcome。这套规范看起来啰嗦但一旦规模变大统一的字段命名会让规则编写和维护容易得多。5.3 日志保留周期怎么定日志保留周期不是一个技术问题而是一个合规和成本问题。从安全角度至少需要保留 180 天因为很多攻击是慢速、低频的攻击者可能在几个月后才会利用窃取的凭据。从等保合规角度不同级别的系统对日志保留时间有明确要求最常见的标准是不少于 6 个月。实际操作中建议分冷热两层热数据保留最近 7 天支持快速检索和告警温数据保留 180 天用于合规审计和追溯超过 180 天的数据做压缩归档或者转移到对象存储。这个策略可以显著降低存储成本同时满足绝大多数合规要求。5.4 团队协作与权限管控日志里包含大量敏感信息比如用户名、IP、SQL 语句、文件路径。如果所有开发都能看到全部日志会产生数据泄露风险。建议利用 Kibana 的 Space 和 Role 功能做权限隔离。例如安全团队拥有全部索引的读写权开发团队只能查看业务日志索引运维团队只能查看系统日志索引。具体做法是在 Stack Management Roles 里创建不同角色在 Index Privileges 里指定允许访问的索引模式和操作权限然后在 Kibana Spaces 里为不同团队分配空间。这样每个团队打开 Kibana只能看到自己被授权的数据和仪表板。权限管控看似繁琐但在企业环境里迟早要做越早做越好。5.5 告警疲劳的治理思路告警疲劳是安全运营里最隐蔽的杀手。告警太多安全人员会逐渐忽略所有告警真正严重的事件反而被发现不了。治理思路有三步第一基线化。先跑两周数据看正常波动范围再设定阈值。第二分级。把告警分成紧急、高、中、低四档紧急告警直接打电话或发短信低级告警只在日报里汇总。第三闭环。每一条告警都必须有处理状态已确认、处理中、已解决、误报。没有闭环的告警体系很快就会失去价值。在 ELK 里实现分级比较简单可以在告警通知内容里带不同的标签字段或者创建多个规则不同阈值对应不同通知渠道。关键是运营同学要定期复盘告警清单逐步调优规则把误报率降下来。6. 踩坑后的几点心得做到这一步你会发现 ELK 做安全可观测最花时间的不是部署和配置而是对数据源的理解和清洗规则的打磨。我在实战中最深的体会是先别急着追求漂亮的仪表板先把一条链路的日志从采集、解析、存储到查询完整打通比什么都重要。一条链路走通了后面复制到其他数据源就是体力活。还有一个小技巧每次新增一种日志类型先手工采集 100 条左右的样本日志仔细分析格式再写解析规则。不要拿生产数据直接在 Grok 上试容易把错误规则发布上去导致线上日志解析失败。用样本调试好再发布到生产出错概率会小很多。另外Logstash 配置改动后一定要先本地测试再重启。可以用logstash -f /path/to/config --config.test_and_exit检查配置语法是否错误避免一把梭重启导致数据断流。生产环境改动解析规则时还要考虑已经在 pipeline 里的数据如何处理是重放还是接受字段缺失最好提前想清楚。后续我会在这个系列里继续写如何设计一套完整的 ECS 字段映射规范、如何把威胁情报源接入 ELK 做 IP 信誉查询、如何用 Elastic Security 的检测规则覆盖常见攻击场景、如何做安全日志的自动化报表与周报。如果你在搭建过程中遇到什么问题欢迎带着报错信息和配置片段来讨论很多坑不是我在这里写一遍就能完全覆盖的需要结合实际环境去定位。日志这条线做好之后你会明显感觉到安全事件处理从“大海捞针”变成“按图索骥”这才是这一整套体系最大的价值。
返回列表