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, 力导向图, 威胁情报, 安全剧本, 安全防御评估, 开发者工具, 开源安全工具, 异步编程, 搜索引擎查询, 无后门, 日志处理, 网络安全, 自动化运维, 请求拦截, 逆向工具, 逆向工程平台, 隐私保护