本文完整演示：漏洞复现 → 攻击流量采集 → 特征检测 → 可视化溯源 → 告警分级 → 自动封禁 → 审计留痕，基于 ELK+Filebeat+Python 搭建小型安全运营平台，适合安全学习、靶场实践。
⚠️ 免责声明：本项目仅用于安全学习与授权测试，所有攻击模拟均在自建虚拟机环境中完成。请勿用于任何未授权的攻击行为，一切后果由使用者自负。

前言

很多安全同学在做完漏洞复现后，就止步于 “弹个计算器、拿个 shell”，却忽略了一个更重要的问题：这个攻击行为，如何在企业内被第一时间发现、告警、并自动响应？

本文以 CVE-2024-57521（RuoYi v4.7.9 认证后 SQL 注入）为切入点，搭建一套基于 ELK Stack + Filebeat + Python 的安全日志分析与自动化告警响应平台，完整实现安全运营闭环。

搭建过程中踩了 10+ 个真实环境坑（Docker 镜像白名单、apt 锁冲突、容器网络隔离、ES X-Pack 401、GeoIP 类型不兼容等），全部记录排查与解决方案。

测试环境清单

表格

1. 漏洞背景：CVE-2024-57521

1.1 漏洞简介

RuoYi 是一款基于 Spring Boot 的国产后台管理框架。

• 漏洞类型：认证后 SQL 注入
• 影响版本：RuoYi v4.7.9 及以下
• 利用条件：需要登录后台，具备基础访问权限
• 危害：读取数据库敏感信息、绕过后台权限控制

1.2 漏洞成因

部分接口未对用户输入参数做参数化查询，直接拼接 SQL 语句执行。攻击者可传入 UNION SELECT 等语句读取任意表数据。

1.3 复现环境搭建：部署 RuoYi-Vue

cd /opt
sudo git clone https://gitee.com/y_project/RuoYi-Vue.git
cd RuoYi-Vue

# 初始化数据库
mysql -uroot -p -e "CREATE DATABASE `ry-vue` CHARACTER SET utf8mb4"
mysql -uroot -p ry-vue < sql/ry_20260417.sql
mysql -uroot -p ry-vue < sql/quartz.sql

# 修改数据库配置
nano ruoyi-admin/src/main/resources/application-druid.yml
# username: ruoyi / password: Liao0420

# 打包并启动
mvn clean package -DskipTests
nohup java -jar ruoyi-admin/target/ruoyi-admin.jar > ruoyi.log 2>&1 &

1.4 漏洞验证

登录后台后，携带恶意参数发送请求：

curl "http://192.168.220.132/prod-api/system/user/list?username=1%27%20UNION%20SELECT%201,2,3--+"

Payload 解码：

username=1' UNION SELECT 1,2,3--+

攻击成功，服务器返回数据库敏感字段内容。

核心问题：如果是生产服务器，管理员如何发现攻击？这就是本项目要解决的安全运营闭环。

2. 技术架构设计

2.1 整体架构

2.2 技术选型

表格

​

3. ELK 部署踩坑实录

⚠️ 踩坑一：Ubuntu 后台自动更新抢占 apt 锁

安装 Docker 报错：

E: Could not get lock /var/lib/dpkg/lock-frontend

原因：Ubuntu 默认开启 unattended-upgrades，后台自动更新抢占 apt 锁。

解决：

sudo systemctl stop unattended-upgrades
sudo systemctl disable unattended-upgrades
sudo killall apt apt-get dpkg
sudo rm -f /var/lib/dpkg/lock-frontend /var/lib/dpkg/lock /var/cache/apt/archives/lock
sudo dpkg --configure -a

⚠️ 踩坑二：Docker 镜像被加速器白名单拦截

拉取 jertel/elastalert2 报错：

ERROR: error from registry: this image is not in the allowlist

解决：更换国内镜像加速源

sudo tee /etc/docker/daemon.json << 'EOF'
{
  "registry-mirrors": [
    "https://docker.1panel.live",
    "https://hub.rat.dev"
  ]
}
EOF
sudo systemctl daemon-reload
sudo systemctl restart docker

3.3 docker-compose.yml

services:
  es:
    image: docker.elastic.co/elasticsearch/elasticsearch:7.17.10
    container_name: es
    environment:
      - discovery.type=single-node
      - ES_JAVA_OPTS=-Xms512m -Xmx512m
      - xpack.security.enabled=false
    ulimits:
      memlock: { soft: -1, hard: -1 }
    volumes:
      - ./es-data:/usr/share/elasticsearch/data
    ports:
      - "9200:9200"
    networks:
      - elk
  kibana:
    image: docker.elastic.co/kibana/kibana:7.17.10
    container_name: kibana
    environment:
      - ELASTICSEARCH_HOSTS=http://es:9200
    ports:
      - "5601:5601"
    depends_on:
      - es
    networks:
      - elk
  logstash:
    image: docker.elastic.co/logstash/logstash:7.17.10
    container_name: logstash
    volumes:
      - ./logstash/logstash.conf:/usr/share/logstash/pipeline/logstash.conf:ro
    ports:
      - "5044:5044"
    depends_on:
      - es
    networks:
      - elk
networks:
  elk:

启动命令：

sudo docker compose up -d
# 验证ES
curl http://localhost:9200

返回 "tagline": "You Know, for Search" 代表 ES 启动成功。

⚠️ 踩坑三：Logstash 报 No route to host

报错：

No route to host (Host unreachable) url: http://es:9200

原因：手动 docker start/restart 容器不会自动加入 Compose 自定义网络。
解决：

cd /opt/secops-elk
sudo docker compose down
sudo docker compose up -d

⚠️ 踩坑四：ES X-Pack 返回 401

Got response code '401' contacting Elasticsearch at URL 'http://es:9200/'

原因：重建容器时 xpack.security.enabled 被误改为 true。
解决：

sed -i 's/xpack.security.enabled=true/xpack.security.enabled=false/' docker-compose.yml
sudo docker compose up -d es

4. Filebeat 日志采集

4.1 filebeat.yml

filebeat.inputs:
- type: log
  enabled: true
  paths:
    - /var/log/nginx/access.log
  fields:
    log_type: nginx_access
- type: log
  enabled: true
  paths:
    - /var/log/auth.log
  fields:
    log_type: ssh_auth
- type: log
  enabled: true
  paths:
    - /var/log/syslog
  fields:
    log_type: syslog
output.logstash:
  hosts: ["localhost:5044"]

⚠️ 踩坑五：Filebeat 无权限读取 auth.log

解决：

sudo chmod 644 /var/log/auth.log
sudo systemctl restart filebeat

5. Logstash 攻击特征检测

5.1 logstash.conf

input {
  beats { port => 5044 }
}
filter {
  # ========== Nginx 访问日志解析 ==========
  if [fields][log_type] == "nginx_access" {
    grok {
      match => {
        "message" => "%{IPORHOST:client_ip} - %{DATA:ident} \[%{HTTPDATE:timestamp}\] \"%{WORD:method} %{DATA:request} HTTP/%{NUMBER:http_version}\" %{NUMBER:status:int} %{NUMBER:body_bytes_sent:int} \"%{DATA:referrer}\" \"%{DATA:user_agent}\""
      }
    }
    date {
      match => ["timestamp", "dd/MMM/yyyy:HH:mm:ss Z"]
      target => "@timestamp"
    }
    # SQL 注入特征检测规则
    if [request] =~ /(?i)(union.*select|select.*from|sleep\(|benchmark\(|information_schema|or\s+1=1|'.*or.*')/ {
      mutate { add_field => { "attack_type" => "sql_injection" } }
    }
  }
  # ========== SSH 暴力破解检测 ==========
  if "Failed password" in [message] {
    mutate { add_field => { "attack_type" => "ssh_bruteforce" } }
    grok {
      match => { "message" => "from %{IP:src_ip}" }
    }
    geoip {
      source => "src_ip"
      target => "geoip"
    }
  }
}
output {
  elasticsearch {
    hosts => ["http://es:9200"]
    index => "secops-%{+YYYY.MM.dd}"
  }
}

⚠️ 踩坑六：grokparsefailure

原因：原始日志中 invalid user wronguser，贪婪正则匹配失败。
解决：简化 grok 表达式，只提取 src_ip：

"from %{IP:src_ip}"

⚠️ 踩坑七：GeoIP 的 geo_point 类型不兼容

"caused_by": { "type": "number_format_exception", "reason": "For input string: \"%{[geo][latitude]}\"" }

原因：Logstash7.17 输出经纬度为字符串，ES geo_point 需要数字类型，转换报错。
解决：放弃 geo_point，改用 geoip.country_name 做攻击来源国家统计。

​

6. Python 自动化告警与封禁

6.1 告警分级策略

统计窗口：1 分钟内失败次数

表格

附加策略

• 冷却：同一 IP 10 分钟内只告警一次，防止告警风暴
• 白名单：内网、网关、管理 IP 永不封禁，避免误封
• 审计：所有告警 / 封禁动作写入 audit.log

6.2 核心脚本（alarm.py 节选）

import time
import subprocess

last_alert_time = dict()
WHITELIST = ["127.0.0.1", "192.168.220.1"]

def check_ssh_bruteforce():
    query = {
        "query": {
            "bool": {
                "must": [
                    {"match_phrase": {"message": "Failed password"}},
                    {"range": {"@timestamp": {"gte": "now-1m"}}}
                ]
            }
        },
        "aggs": {
            "by_src_ip": {"terms": {"field": "src_ip.keyword", "size": 20}}
        }
    }
    # ES 查询逻辑省略
    for b in buckets:
        ip, count = b["key"], b["doc_count"]
        level = "高危" if count >= 20 else "中危" if count >= 10 else "低危" if count >= 5 else None
        if not level:
            continue
        if time.time() - last_alert_time.get(ip, 0) < 600:
            continue
        send_dingtalk_alert(f"IP: {ip}, 次数: {count}", level)
        if count >= 20:
            block_ip(ip)

def block_ip(ip):
    if ip in WHITELIST:
        audit_log("SKIP", ip, "命中白名单")
        return
    subprocess.run(["sudo", "iptables", "-I", "INPUT", "-s", ip, "-j", "DROP"])
    audit_log("BLOCK", ip, "已封禁")

6.3 配置 sudo 免密执行 iptables

echo "$USER ALL=(ALL) NOPASSWD: /usr/sbin/iptables" | sudo tee /etc/sudoers.d/secops-iptables
sudo chmod 440 /etc/sudoers.d/secops-iptables

6.4 后台常驻运行告警脚本

cd /opt/secops-elk/scripts
nohup python3 -u alarm.py > alarm.log 2>&1 &

7. 攻击验证与闭环演示

7.1 SQL 注入检测验证

发送 payload：

curl "http://192.168.220.132/prod-api/system/user/list?username=1%27%20UNION%20SELECT%201,2,3--+"

Kibana Discover 检索：attack_type: sql_injection
成功捕获字段：

client_ip: 192.168.220.1
request: /prod-api/system/user/list?username=1%27%20UNION%20SELECT%201,2,3--+
attack_type: sql_injection

7.2 SSH 暴力破解 + 告警分级

模拟 25 次密码错误：

for i in {1..25}; do
  sshpass -p "wrongpass" ssh -o StrictHostKeyChecking=no wronguser@localhost
done

alarm.log 输出示例：

[*] 攻击源 IP: 127.0.0.1, 失败次数: 14
[+] 钉钉告警发送成功，状态码: 200
[2026-10-15 17:42:10] [ALERT] IP=127.0.0.1 级别=高危, 次数=14
[2026-10-15 17:42:10] [BLOCK] IP=127.0.0.1 已通过 iptables 封禁

7.3 自动封禁验证

sudo iptables -L INPUT -n --line-numbers | grep DROP

输出示例：

DROP  all  --  127.0.0.1  0.0.0.0/0

攻击源 IP 被自动封禁，安全运营闭环完成。

8. Kibana 可视化大屏

表格

9. 踩坑总结表

表格

10. 未来优化方向

• 采集 Windows 事件日志（Winlogbeat）
• 集成威胁情报库（AlienVault OTX、微步）
• MITRE ATT&CK 战术映射
• 告警对接企业微信 / 飞书
• Kibana ML 异常检测
• ES 多节点集群 + ILM 日志生命周期管理

参考链接

• Elastic 官方文档：https://www.elastic.co/guide/
• RuoYi 项目：https://gitee.com/y_project/RuoYi-Vue
• CVE-2024-57521：https://nvd.nist.gov/vuln/detail/CVE-2024-57521

GitHub 配套文件建议（额外给你）

你可以在仓库根目录新建：

1. README.md：直接粘贴上面这份（就是主页展示）
2. docker-compose.yml：单独放 compose 配置
3. logstash/logstash.conf：单独放 logstash 管道配置
4. filebeat/filebeat.yml
5. scripts/alarm.py：告警脚本
6. imgs/：存放截图，文中 【图xx】 改成 ![图描述](./imgs/xxx.png)
7. .gitignore：忽略 es-data、日志、nohup 输出

# es数据目录
es-data/
# 日志文件
*.log
nohup.out
# python缓存
__pycache__/
*.pyc