Hu9o73/elasticsearch-data-generator
GitHub: Hu9o73/elasticsearch-data-generator
该项目为 Elasticsearch 批量生成符合 ECS 格式并映射 MITRE ATT&CK 的逼真安全事件数据，用于测试和验证 SIEM/SOC 平台的检测能力。
Stars: 0 | Forks: 0
# Elasticsearch 数据生成器
**一个用于 Elasticsearch 的逼真安全数据生成器**
该项目旨在生成逼真的安全事件并将其注入到 Elasticsearch 中，以测试 SIEM、SOC 和安全分析解决方案（如 FusionAI）。
改编自 [splunk-data-generator](https://github.com/philoo99999/splunk-data-generator)，以适配 Elasticsearch 8.x。
## 🎯 特性
- ✅ 生成 500 MB 至 3+ GB 的逼真攻击数据
- ✅ 支持多种网络攻击类型（SQL Injection、XSS、Lateral Movement、Data Exfiltration 等）
- ✅ MITRE ATT&CK 映射（技术与战术）
- ✅ 兼容 ECS (Elastic Common Schema) 格式
- ✅ 集成 Active Directory（用户）和 CMDB（资产）
- ✅ 通过 Elasticsearch Bulk API 注入
- ✅ 兼容 Elasticsearch 8.x
## 🔐 生成的攻击类型
| 攻击 | 严重性 | MITRE 技术 | 描述 |
|---------|----------|------------------|-------------|
| **SQL Injection** | CRITICAL | T1190, T1189 | 包含逼真 SQL 注入的恶意 URL |
| **Cross-Site Scripting (XSS)** | HIGH | T1189, T1203 | 注入的 JavaScript 脚本，多种 XSS Payload |
| **Lateral Movement** | CRITICAL | T1021, T1550 | SMB、RDP、Pass-the-Hash，可疑的网络移动 |
| **Data Exfiltration** | CRITICAL | T1048, T1041 | DNS 数据外泄，大规模 FTP/HTTPS 传输 |
| **Reconnaissance** | MEDIUM | T1046, T1087 | 端口扫描，网络和 LDAP 枚举 |
## 📋 前置条件
- Python 3.6+
- 正在运行的 Elasticsearch 8.x
- Python 库：`requests`、`urllib3`、`elasticsearch`
## 📂 项目结构
```
elasticsearch-data-generator/
├── scripts/
│ ├── generate_events.py # Génère les événements
│ ├── inject_to_es.py # Injecte dans Elasticsearch
│ └── verify_es.py # Vérifie les données injectées
├── requirements.txt
├── .gitignore
└── README.md
```
## 🚀 安装
1. **克隆此 repository：**
```
git clone https://github.com/philoo99999/elasticsearch-data-generator.git
cd elasticsearch-data-generator
```
2. **安装依赖：**
```
pip install -r requirements.txt
```
3. **准备您的参考数据：**
- `DATABASE_FusionAI.db` - 包含告警和 IP 的 SQLite 数据库
- `ad_users.csv`（可选）- Active Directory 用户列表
- `cmdb_assets.csv`（可选）- 您 CMDB 中的资产列表
## ⚡ 快速开始（attack-chain generator）
- 将 `scripts/.env.example` 复制到 `scripts/.env` 并调整路径（`DB_PATH`、`AD_USERS_FILE`、`CMDB_ASSETS_FILE`、`OUTPUT_PREFIX`）。
- （可选）设置种子以确保运行结果可复现：export `SEED=42`。
- 生成关联的攻击链 + 逼真的噪声数据：
python3 scripts/generate_attack_chains.py
- NDJSON 文件将写入 `OUTPUT_PREFIX####.json`；JSON 摘要将生成至 `SUMMARY_PATH`（或默认的 `_summary.json`）。
## 💻 使用说明
### 1. 启动 Elasticsearch
确保 Elasticsearch 正在运行：
```
sudo systemctl status elasticsearch
```
### 2. 生成数据
生成 500 MB 的安全事件数据：
```
python3 scripts/generate_events.py
```
该脚本将生成：
- 700,000+ 条事件
- 7 个 batch 文件（每个约 71 MB）
- 跨越 30 天分布的事件
- 混合逼真的攻击并进行 MITRE ATT&CK 映射
**自定义：**
修改脚本中的 `TARGET_SIZE_MB` 以生成更多或更少的数据。
### 3. 注入 Elasticsearch
**配置：**
使用您的 Elasticsearch credentials 编辑 `scripts/inject_to_es.py`：
```
ES_URL = "https://localhost:9200"
ES_USER = "elastic"
ES_PASS = "votre_mot_de_passe"
```
**注入：**
```
python3 scripts/inject_to_es.py
```
该脚本将：
- 连接到 Elasticsearch
- 创建带有 ECS mapping 的 index template
- 通过 Bulk API 注入事件
- 实时显示进度
- 预计时间：500 MB 需 5-10 分钟
### 4. 验证注入
```
python3 scripts/verify_es.py
```
或直接使用 curl：
```
# 统计事件
curl -k -u elastic:PASSWORD https://localhost:9200/fusionai-*/_count
# 查看 index
curl -k -u elastic:PASSWORD https://localhost:9200/_cat/indices/fusionai-*?v
# 搜索
curl -k -u elastic:PASSWORD https://localhost:9200/fusionai-*/_search?size=5
```
## 📊 Elasticsearch 查询示例
### Top 10 源 IP
```
GET /fusionai-*/_search
{
"size": 0,
"aggs": {
"top_sources": {
"terms": {
"field": "source.ip",
"size": 10
}
}
}
}
```
### 按严重性划分的攻击
```
GET /fusionai-*/_search
{
"size": 0,
"aggs": {
"by_severity": {
"terms": {
"field": "event.severity"
}
}
}
}
```
### 攻击时间线
```
GET /fusionai-*/_search
{
"size": 0,
"aggs": {
"timeline": {
"date_histogram": {
"field": "@timestamp",
"calendar_interval": "day"
},
"aggs": {
"by_category": {
"terms": {
"field": "security.category"
}
}
}
}
}
}
```
### MITRE ATT&CK 技术
```
GET /fusionai-*/_search
{
"size": 0,
"aggs": {
"mitre_techniques": {
"terms": {
"field": "threat.technique.id",
"size": 20
},
"aggs": {
"tactics": {
"terms": {
"field": "threat.tactic.name"
}
}
}
}
}
}
```
## 🧪 测试
- 单元测试通过 `unittest` 验证生成器的一些不变量（比率边界、时间戳 clamp、host/destination 一致性）。
- 执行：
python -m unittest tests/test_generator.py
如有需要，请将项目添加到 `PYTHONPATH` 中；`tests/test_generator.py` 会自动从根目录插入它。
### 特定 IP 的事件
```
GET /fusionai-*/_search
{
"query": {
"term": {
"source.ip": "10.14.45.103"
}
}
}
```
## 🔧 配置
### ECS (Elastic Common Schema) 格式
生成的事件遵循 ECS 格式，以最大程度地兼容 Elastic 生态系统：
```
{
"@timestamp": "2025-11-16T10:30:00.000Z",
"event": {
"category": "security",
"severity": "critical",
"kind": "alert"
},
"source": {
"ip": "192.168.1.100",
"port": 54321
},
"destination": {
"ip": "10.0.1.50",
"port": 443
},
"threat": {
"framework": "MITRE ATT&CK",
"technique": {
"id": "T1190",
"name": "SQL Injection"
}
}
}
```
### 参考数据
#### ad_users.csv 格式（可选）
```
Username,Department,Display_Name
jsmith,IT,John Smith
mdoe,Finance,Mary Doe
```
#### cmdb_assets.csv 格式（可选）
```
Hostname,Asset_Type,Criticality,Location
WKS-001,Workstation,Medium,Building A
SRV-001,Server,Critical,DataCenter
```
如果未提供这些文件，生成器将创建默认数据。
## 🔍 故障排除
### 错误：“Connection refused”
Elasticsearch 未启动：
```
sudo systemctl start elasticsearch
sudo systemctl status elasticsearch
```
### 错误：“Authentication failed”
检查您在 `inject_to_es.py` 中的 credentials：
```
# 重置密码
sudo /usr/share/elasticsearch/bin/elasticsearch-reset-password -u elastic
```
### 错误：“Index creation failed”
检查您是否拥有足够的权限以及该 index 是否已存在：
```
curl -k -u elastic:PASSWORD -X DELETE https://localhost:9200/fusionai-*
```
## 🎨 使用 Kibana 进行可视化
1. **安装 Kibana**（可选）：
```
sudo apt-get install kibana
sudo systemctl start kibana
```
2. **访问 Kibana**：http://localhost:5601
3. **创建 Data View**：
- Stack Management → Data Views
- Create data view：`fusionai-*`
- Timestamp field：`@timestamp`
4. **创建可视化**：
- Dashboard → Create visualization
- 使用 ECS 字段创建图表
## 📈 性能
| 指标 | 值 |
|----------|--------|
| 事件大小 | ~800 bytes/事件 |
| 生成速度 | ~50,000 事件/s |
| 注入速度 | ~5,000-10,000 事件/s |
| 500 MB 耗时 | ~5-10 分钟 |
## 🔒 安全
- ⚠️ 切勿公开分享您的 Elasticsearch 密码
- ⚠️ 生成的数据仅供测试环境使用
- ⚠️ 在生产环境中使用 HTTPS 和强密码
## 📄 许可证
MIT License - 有关更多详细信息，请参阅 LICENSE 文件
## 👤 作者
从 [splunk-data-generator](https://github.com/philoo99999/splunk-data-generator) 改编以适配 Elasticsearch
为测试和开发 SIEM/SOC 解决方案（如 FusionAI）而创建的项目。
## 🔗 相关链接
- [Elasticsearch 文档](https://www.elastic.co/guide/en/elasticsearch/reference/current/index.html)
- [Elastic Common Schema (ECS)](https://www.elastic.co/guide/en/ecs/current/index.html)
- [MITRE ATT&CK 框架](https://attack.mitre.org/)
- [Kibana 用户指南](https://www.elastic.co/guide/en/kibana/current/index.html)
## 📝 更新日志
### v1.0.0 (2025-11-16)
- ✅ 初始生成 500 MB 数据
- ✅ 支持 5 种主要攻击类型
- ✅ 兼容 ECS 格式
- ✅ 通过 Bulk API 注入
- ✅ 验证脚本
- ✅ 完整的文档
**注意：** 本项目专为测试和开发环境设计。未经适当验证，请勿在生产环境中使用。
标签：CISA项目, Cloudflare, Elasticsearch, Go语言工具, MITRE ATT&CK, Modbus, PE 加载器, 安全测试, 插件系统, 攻击性安全, 数据生成, 越狱测试, 逆向工具