Akeju-ak/soc-hunt-engine

# SOC Hunt Engine — 多源威胁检测流水线 一个生产级的安全运营中心（SOC）流水线，使用 **Python** 和 **DuckDB** 构建。该流水线能够规范化多源安全遥测数据，将可疑活动与变更管理记录进行比对排查，并在统一的时间线上重建多阶段攻击战役。 ## 项目背景 SOC 无法直接获得干净的数据。同一个字段在不同日志版本中会以不同的名称出现（schema drift），不同系统之间的时钟同步会发生偏移，并且网络重传会产生重复的事件。本项目将上述问题视为必须首先解决的实际核心难题——在任何检测规则编写之前——通过构建一个能够摄取、规范化、验证原始遥测数据并将其关联起来的流水线，使其成为人类分析师真正能够调查的线索。 ## 功能介绍 该流水线分三个阶段运行： ### 1. 摄取与规范化 (`hunt-engine/ingest.py`, `hunt-engine/normalize.py`) 将来自 5 个遥测源（**Auth、Web、DNS、Firewall 和 EDR**）的安全事件加载到统一的 DuckDB schema（`norm.events`）中。Auth 规范化器专门处理了 3 种不同的 schema 版本（timestamp、identity 和 action 的字段名各不相同），并且任何无法解析或缺少必填字段的记录都会被打上原因代码被路由到 `raw.quarantine` 中，而不会被直接静默丢弃。 ### 2. 告警排查 (`hunt-engine/triage.py`) 通过严格的 **7 项合规性检查**，将 96 个被标记的活动候选者与企业变更管理工单进行交叉比对： | # | 检查项 | 要求 | |---|-------|-------------| | 1 | 工单是否存在 | `activityId` 必须有匹配的变更记录 | | 2 | 状态 | 必须为 `APPROVED` | | 3 | 审批权限 | 必须为 `change-board` | | 4 | 资产范围 | 必须在 `asset-001`–`asset-050` 范围内 | | 5 | 操作者授权 | 必须在 `svc-001`–`svc-050` 范围内 | | 6 | 部门归属 | 安全工程部、平台部、数据服务部或 IT 运营部 | | 7 | 时间窗口 | 事件必须发生在工单时间窗口内，持续时间 ≤ 2 小时 | 只有当 **7 项检查全部通过** 时，活动才会被确认为良性。结果：**80 个确认为误报**，**16 个升级为未授权安全事件**。 ### 3. 战役重建 (`hunt-engine/campaigns.py`) 在单一的 UTC 时间线上关联所有 5 个来源的规范化事件，以识别完整的攻击故事线，而不是孤立的告警。系统重建了三个战役： - **凭证填充与横向移动** — 登录失败激增 → 登录成功 → 执行混淆的 PowerShell - **恶意 Web Payload 与 C2** — HTTP 文件下载 → C2 域名查询 → 防火墙出站阻断 - **权限提升与持久化** — LSASS 凭证转储 → 创建计划任务 → 尝试清除审计日志 ## 架构 ``` Raw Telemetry (Auth / Web / DNS / Firewall / EDR) │ ▼ normalize.py ──► norm.events (clean) ──► raw.quarantine (malformed/invalid) │ ▼ triage.py ──► outputs/triage_results.json ──► outputs/tp-fp-table.csv │ ▼ campaigns.py ──► outputs/normalized-timeline.csv ──► outputs/campaign-graph.json ``` ## 项目结构 ``` . ├── hunt-engine/ │ ├── ingest.py # Loads events into DuckDB │ ├── normalize.py # Schema-drift normalization + quarantine logic │ ├── triage.py # 7-point compliance triage engine │ └── campaigns.py # Cross-source campaign correlation ├── queries/ │ └── schema.sql # DuckDB table definitions ├── data/ │ └── ubi-2026-0083-stage-5-discrepancy.json # Alert candidates + change records ├── outputs/ │ ├── triage_results.json │ ├── tp-fp-table.csv │ ├── normalized-timeline.csv │ └── campaign-graph.json ├── tests/ │ └── test_pipeline.py # Automated verification of triage + campaign counts ├── evidence-index.csv ├── manifest.sha256 # Integrity hashes for all deliverables └── README.md ``` ## 快速开始 **环境要求：** Python 3.10+, pip ``` # 1. Clone the repo git clone https://github.com/Akeju-ak/soc-hunt-engine.git cd soc-hunt-engine # 2. 设置 virtual environment python3 -m venv venv source venv/bin/activate # On Windows: venv\Scripts\activate # 3. 安装 dependencies pip install --upgrade pip pip install duckdb pytest # 4. 运行 pipeline python3 hunt-engine/ingest.py # Ingest & normalize telemetry python3 hunt-engine/triage.py # Run the 7-point triage engine python3 hunt-engine/campaigns.py # Reconstruct threat campaigns # 5. 使用 automated test suite 验证一切 python3 -m pytest tests/ ``` ## 验证与完整性 - `tests/test_pipeline.py` 会断言确切的排查计数（96 个已评估，80 个良性，16 个已升级）和战役数量（3 个），因此逻辑中的任何回归都会导致构建立即失败。 - `manifest.sha256` 包含了每个交付文件的 SHA-256 哈希值，用于完整性验证。 - 所有输出均通过 `data/ubi-2026-0083-stage-5-discrepancy.json` 以编程方式生成 — `outputs/` 中的任何内容都没有经过手动编辑。 ## 技术栈 Python · DuckDB · pytest · JSON/CSV 结构化数据 · 符合 MITRE ATT&CK 标准的战役映射 ## 许可证 如果您打算将此代码库公开并供他人复用，请添加您选择的许可证（例如 MIT）。

标签：DuckDB, Homebrew安装, Python, 安全数据处理, 安全规则引擎, 安全运营, 扫描框架, 无后门, 逆向工具