lord-dubious/ai-enhanced-siem

# AI 增强的 SIEM 一个安全信息和事件管理 (SIEM) pipeline，使用 `msgspec` 解析 Wazuh 警报，在可用时使用 Redis 进行去重/存储，并可以调用 Gemini 进行可选的警报富化分析。当 Redis 或 Gemini 不可用时，pipeline 会在警报和缓存统计信息上记录降级模式的元数据，而不是隐藏故障。 ## 作品集展示  - **架构深入解析：** [`docs/ARCHITECTURE.md`](docs/ARCHITECTURE.md) - **演示指南：** [`docs/DEMO.md`](docs/DEMO.md) - **审查重点：** 类型化警报接收、Redis 降级状态跟踪、Gemini 富化分析以及人工审查的响应建议。 ## 架构概述 ``` flowchart TB classDef input fill:#ecfeff,stroke:#0891b2,stroke-width:2px,color:#164e63 classDef core fill:#eef2ff,stroke:#4f46e5,stroke-width:2px,color:#312e81 classDef external fill:#fff7ed,stroke:#ea580c,stroke-width:2px,color:#7c2d12 classDef metadata fill:#f0fdf4,stroke:#16a34a,stroke-width:2px,color:#14532d classDef review fill:#fef2f2,stroke:#dc2626,stroke-width:2px,color:#7f1d1d Alerts[/Wazuh-style alert JSON/]:::input Analyst[/SOC analyst review/]:::review subgraph Intake["Typed Intake"] Parser[msgspec parser]:::core Batch[Alert batch model]:::metadata end subgraph State["Cache and Deduplication Boundary"] Cache[Redis cache manager]:::core Redis[(Redis optional)]:::external CacheStatus[cache_status cache_error degraded state]:::metadata end subgraph Triage["Enrichment Boundary"] Enricher[Gemini alert enricher]:::core Gemini{{Gemini API optional}}:::external EnrichmentStatus[enrichment source status error]:::metadata end subgraph Response["Human-Gated Response"] Firewall[Firewall rule suggestions]:::core Warnings[Batch warnings]:::metadata Output[Enriched alerts and suggestions]:::review end Alerts --> Parser --> Batch --> Cache Cache <-->|dedupe state| Redis Cache -. Redis unavailable .-> CacheStatus Cache --> Enricher Enricher <-->|optional triage| Gemini Enricher -. unavailable or failed .-> EnrichmentStatus Enricher --> Firewall Firewall -->|requires human review| Output CacheStatus --> Warnings EnrichmentStatus --> Warnings Warnings --> Output --> Analyst ``` ## 功能 - **结构化解析**：使用 `msgspec` 对 Wazuh 警报进行 JSON 反序列化 - **Redis 去重**：基于 Redis 的警报去重，具有显式的降级状态报告 - **可选的 Gemini 富化分析**：配置后，Gemini 可以将警报映射到 MITRE ATT&CK 战术并建议补救措施 - **流处理**：对 Wazuh `alerts.json` 进行流式处理 - **IOC 提取**：自动提取 IP 地址、域名、文件哈希和其他指标 - **防火墙规则建议**：建议的防火墙规则包含来源元数据，在使用前需要人工审查 ## 安装 ### 前置条件 - Python 3.11+ - Redis（用于生产环境去重） - Wazuh Manager（用于警报接收） - Gemini API 密钥 ### 快速开始 ``` # 克隆 repository git clone https://github.com/lord-dubious/ai-enhanced-siem.git cd ai-enhanced-siem # 创建 virtual environment uv venv source .venv/bin/activate # 安装 dependencies uv pip install -e ".[dev]" # 设置 environment variables cp .env.example .env # 使用你的 GEMINI_API_KEY 编辑 .env # 运行测试 pytest tests/ -v ``` ### Docker 部署 ``` # 启动 full stack (Redis + Pipeline) docker-compose up -d # 或者使用 Wazuh 和 OpenSearch docker-compose --profile full-stack up -d ``` ## 使用方法 ### CLI 命令 ``` # 处理单个 alert 文件 ai-siem process alerts.json # 实时监控并处理 alerts ai-siem process --watch /var/ossec/logs/alerts/alerts.json # 使用自定义 output 进行处理 ai-siem process alerts.json --output enriched-alerts.json # 分析特定 alert ai-siem analyze '{"rule": {"id": "100001", "level": 12, ...}}' ``` ### Python API ``` from ai_siem import Pipeline, WazuhAlertParser, AlertCache, AIEnricher # 初始化 components parser = WazuhAlertParser() cache = AlertCache(redis_url="redis://localhost:6379") enricher = AIEnricher(api_key="your-gemini-api-key") # 创建 pipeline pipeline = Pipeline(parser=parser, cache=cache, enricher=enricher) # 处理 alerts with open("alerts.json") as f: for line in f: result = await pipeline.process_alert(line) if result: print(f"Enriched: {result.mitre_tactics}") ``` ## 配置 | 环境变量 | 描述 | 默认值 | |---------------------|-------------|---------| | `GEMINI_API_KEY` | Google Gemini API 密钥 | 必填 | | `REDIS_URL` | Redis 连接 URL | `redis://localhost:6379` | | `CACHE_TTL` | 警报去重 TTL（秒） | `3600` | | `LOG_LEVEL` | 日志级别 | `INFO` | | `BATCH_SIZE` | 每批次处理的警报数 | `100` | ## 严重性映射 | Wazuh 级别 | 严重性 | 操作 | |-------------|----------|--------| | 12+ | CRITICAL | 立即响应，Gemini 富化分析 | | 9-11 | HIGH | 优先富化分析 | | 6-8 | MEDIUM | 标准富化分析 | | 0-5 | LOW | 仅记录日志，最少的富化分析 | ## MITRE ATT&CK 和规则建议元数据 Gemini 或模拟富化分析可以添加 MITRE ATT&CK 映射和补救建议。请将防火墙规则视为可审查的建议，而不是可直接用于生产环境的命令： ``` { "alert_id": "1234567890.123", "original_rule": "SSH brute force attack", "severity": "HIGH", "mitre_mapping": { "tactics": ["Initial Access", "Credential Access"], "techniques": ["T1110.001 - Password Guessing"], "sub_techniques": ["T1110.001"] }, "iocs": { "ip_addresses": ["192.168.1.100"], "usernames": ["admin", "root"] }, "enrichment_source": "gemini", "enrichment_status": "success", "recommendations": [ "Review SSH access logs", "Validate source IP reputation" ], "suggested_rules": [ { "action": "block", "source_ip": "192.168.1.100", "port": "22", "source": "gemini", "requires_human_review": true, "safety_note": "Suggestion only; validate context before applying." } ] } ``` ## 性能 性能取决于警报结构、Redis 可用性、硬件和 Gemini API 延迟。该项目目前不发布可复现的基准测试；在进行容量评估之前，请先在您的环境中进行测量。 | 指标 | 值 | |--------|-------| | JSON 解析 | 取决于环境 | | 去重检查 | 取决于 Redis/网络 | | Gemini 富化分析 | 取决于 API 和配额 | | 内存使用量 | 取决于警报量 | ## 测试 ``` # 运行所有测试 pytest tests/ -v # 运行 coverage pytest tests/ -v --cov=ai_siem --cov-report=html # 运行特定 test module pytest tests/test_parser.py -v # 运行 performance tests pytest tests/ -v -k "performance" ``` ## 项目结构 ``` ai-enhanced-siem/ ├── src/ai_siem/ │ ├── __init__.py # Package exports │ ├── models.py # Pydantic + msgspec data models │ ├── parser.py # Wazuh alert parser │ ├── cache.py # Redis deduplication cache │ ├── enricher.py # Optional Gemini enrichment engine │ ├── pipeline.py # Main processing orchestrator │ └── cli.py # Typer CLI interface ├── tests/ │ ├── conftest.py # Pytest fixtures │ ├── test_models.py # Model validation tests │ ├── test_parser.py # Parser tests │ ├── test_cache.py # Cache tests │ ├── test_enricher.py # AI enricher tests │ └── test_pipeline.py # Integration tests ├── docker-compose.yml # Full stack deployment ├── Dockerfile # Production container └── pyproject.toml # Project configuration ``` ## 许可证 MIT 许可证 - 有关详细信息，请参见 [LICENSE](LICENSE)。 ## 致谢 - [Wazuh](https://wazuh.com/) - 开源安全监控 - [msgspec](https://jcristharif.com/msgspec/) - 结构化序列化 - [Google Gemini](https://ai.google.dev/) - 可选的富化分析提供商 - [MITRE ATT&CK](https://attack.mitre.org/) - 威胁框架

标签：AI安全, AMSI绕过, Chat Copilot, DLL 劫持, Gemini, JSON解析, LLM安全运营, msgspec, Python, Redis, Wazuh, 信息收集自动化, 告警分类, 告警去重, 告警富化, 大语言模型, 威胁检测, 安全信息和事件管理, 安全告警, 安全管道, 安全运营中心, 容错处理, 搜索引擎查询, 数据解析, 无后门, 缓存降级, 网络安全, 网络映射, 自动化分诊, 请求拦截, 逆向工具, 防火墙规则, 降级状态追踪, 隐私保护