X-Abhishek-X/soar-engine
GitHub: X-Abhishek-X/soar-engine
一个基于 Python 的轻量级异步 SOAR 引擎,利用 FastAPI 和 Celery 实现高并发安全告警的自动化响应与威胁情报查询。
Stars: 0 | Forks: 0
# soar-engine
我拼凑了这个工具来自动化一些事件响应任务。手动处理原始 IDS 警报很快就会让人厌烦,所以这个工具基本上就是监听 webhook,解析出恶意 IP,并对它们运行自动化的 playbook。
主要目标是确保在流量激增时不会丢失警报。因此,webhook 接收器不是在查询外部 API 时阻塞主线程,而是将 payload 投递到 redis 队列中并立即返回 200 OK。然后,后台的 celery workers 异步获取并处理这些任务。
目前,playbook 只是针对违规 IP 检查 virustotal,并向 slack 频道发送警报。不过很容易扩展。
## 技术栈
- **fastapi**: 用于 webhook 接收端点
- **celery & redis**: 处理异步后台队列
- **python**: 用于 playbook 逻辑
## 本地运行
1. **启动 redis**
如果你只想在 docker 中运行,我包含了一个 compose 文件:
docker-compose up -d
2. **设置 python 环境**
python -m venv venv
source venv/bin/activate
pip install -r requirements.txt
3. **api keys**
在根目录创建一个 `.env` 文件。如果你只是想测试它(它会模拟响应),可以跳过这一步,但要实际使用它你需要:
REDIS_URL=redis://localhost:6379/0
VIRUSTOTAL_API_KEY=your_vt_key
SLACK_WEBHOOK_URL=your_slack_webhook
## 启动项目
你需要分别运行 worker 和 api server。
**终端 1 (the worker):**
```
celery -A tasks worker --loglevel=info
```
**终端 2 (the api):**
```
uvicorn main:app --reload --port 8000
```
## 测试
你可以使用 curl 向 webhook 发送一个虚假警报,以观察 playbook 的运行:
```
curl -X POST http://127.0.0.1:8000/api/v1/webhook/alert \
-H "Content-Type: application/json" \
-d '{
"source_ip": "185.156.73.53",
"alert_type": "Suspected SSH Brute Force",
"severity": "high",
"description": "More than 50 failed login attempts in 1 minute."
}'
```
观察 celery 日志,查看它是如何获取任务并运行 VT 富化的。
标签:AV绕过, Celery, Docker, FastAPI, IDS告警, IP查询, Python, Redis, Slack通知, SOAR, Virustotal, Webhook, 力导向图, 威胁情报, 安全剧本, 安全防御评估, 开发者工具, 开源安全工具, 异步编程, 搜索引擎查询, 无后门, 日志处理, 网络安全, 自动化运维, 请求拦截, 逆向工具, 逆向工程平台, 隐私保护