# Threat Hunting M-ATH 目录
**Model-Assisted Threat Hunting (M-ATH)** — 算法驱动的 Cyber Threat Hunting 主题，与 [Splunk PEAK Threat Hunting 框架](https://www.splunk.com/en_us/blog/security/peak-threat-hunting-framework.html)保持一致。
本仓库管理着使用机器学习和统计方法（如分类、聚类、异常检测和时间序列分析）的威胁狩猎场景，旨在发现更简单的方法可能遗漏的线索。
## M-ATH 与 PEAK 框架
M-ATH 是 PEAK 框架（*Prepare, Execute, and Act with Knowledge*）中三种狩猎类型之一。它使用算法为威胁狩猎寻找线索，从而在以下情况实现更高级和实验性的狩猎：
- **更简单的方法不够准确** — 例如，混入合法流量中的基于字典的 DGA 域名
- **可对行为类别（良性/恶意）进行标记** — 从而实现监督分类
- **数据量巨大或难以汇总** — 适合降维和聚类
- **识别准确但分类困难** — 由 analyst-in-the-loop 进行最终决策
```
graph TD
%% Custom Styling Definitions
classDef prepareStyle fill:#E0F7FA,stroke:#00838F,stroke-width:2px,color:#004D40;
classDef executeStyle fill:#E8EAF6,stroke:#283593,stroke-width:2px,color:#1A237E;
classDef actStyle fill:#FBE9E7,stroke:#D84315,stroke-width:2px,color:#5D4037;
classDef nodeStyle fill:#FFFFFF,stroke:#37474F,stroke-width:1px,color:#263238;
subgraph Phase1 ["1. Prepare Phase"]
A[(Telemetry Sources)] -->|Ingest| B[Data Transform & Sanitization]
B -->|Deduplicated & Anonymized| C[Scenario Input Directory]
end
subgraph Phase2 ["2. Execute Phase (M-ATH)"]
C --> D[Scenario Jupyter Notebook / Script]
D -->|Feature Extraction| E[Models & Statistical Methods]
E -->|Classification / Clustering / Anomaly| F[Candidate Leads]
F -->|Shared Detection Logic Scoring| G[Scored & Enriched Leads]
end
subgraph Phase3 ["3. Act Phase"]
G --> H{High-Confidence Findings}
H -->|Analyst Triage & Investigation| I[Incident Response & Remediation]
end
%% Apply Classes
class Phase1 prepareStyle;
class Phase2 executeStyle;
class Phase3 actStyle;
class A,B,C,D,E,F,G,H,I nodeStyle;
```
## 场景（M-ATH 主题）
场景位于 `scenarios/` 目录下组织，并在 `scenarios/catalog.csv` 中提供了目录。示例包括：
| 类别 | 示例 |
|----------|----------|
| **分类** | 字典 DGA 检测、恶意 Base64 payload、LOLBins (LLM) |
| **聚类** | 进程父子链、相似性分析、user-agent 注入 |
| **异常检测** | DNS/URL 异常分析、网络 beaconing (jittered)、罕见进程行为 |
| **时间序列** | C2 beaconing、受损账户、数据外泄 |
有关用例、模型和数据源的完整列表，请参阅 `scenarios/catalog.csv`。
## 检测逻辑（共享评分与富化规则）
`detection_logics/` 包含场景使用的**模块化“检测逻辑”规则**，用于：
- **在发生命中时提高风险评分**
- **添加标准化的原因标签**（字符串标识符），解释*为什么*对其进行评分
这些规则旨在跨多次狩猎重用，因此场景的 notebooks/scripts 可以专注于分析（特征提取、聚类、异常检测、排名），而将常见的“规则命中”委托给此共享包。
### 如何应用检测逻辑
检测逻辑通过该包公开的辅助函数执行：
- `detection_logics.apply_dns_logics(value: str, decoded_value: str) -> (score_delta: int, reasons: list[str])`
- `detection_logics.apply_url_logics(value: str, decoded_value: str) -> (score_delta: int, reasons: list[str])`
每个逻辑都实现了一个简单的契约：
- `apply(value: str, decoded_value: str) -> (score_delta: int, reason_name: str | None)`
- 如果没有命中，则返回 `(0, None)`。
- 如果规则命中，则返回一个正的分数增量和稳定的原因名称。
**输入**
- `value`：原始/未处理的 DNS 或 URL 值
- `decoded_value`：适用时的解码/规范化版本（例如带有 `punycode:` 或 `base64:` 前缀的值）
**输出**
- `score_delta`：候选项分数应增加的数值
- `reasons`：与命中的规则对应的原因名称列表（用于可审计性和分析师分流）
```
flowchart TD
classDef inputStyle fill:#ECEFF1,stroke:#455A64,stroke-width:1.5px;
classDef routerStyle fill:#E8EAF6,stroke:#3F51B5,stroke-width:1.5px;
classDef logicStyle fill:#E8F5E9,stroke:#2E7D32,stroke-width:1.5px;
classDef aggStyle fill:#FFF8E1,stroke:#F57F17,stroke-width:2px;
classDef outStyle fill:#FFEBEE,stroke:#C62828,stroke-width:2px;
Raw[Raw Telemetry Input]:::inputStyle --> Router{Telemetry Context?}:::routerStyle
Router -->|DNS Value| DNS[apply_dns_logics]:::routerStyle
Router -->|URL Value| URL[apply_url_logics]:::routerStyle
subgraph Logics [Modular Detection Rules]
DNS --> DNS_Susp[dns_suspicious_string.py]:::logicStyle
URL --> DNS_Susp
DNS --> VT_Verd[vt_verdict_not_clean.py]:::logicStyle
URL --> VT_Verd
end
DNS_Susp -->|Score Delta & Tag| Agg[Risk Score Aggregator]:::aggStyle
VT_Verd -->|Score Delta & Tag| Agg
Agg --> Output[Enriched Lead with Risk Score & Reason Tags]:::outStyle
```
### 当前的检测逻辑模块
在提交 `62e72c5b475a51addb1a843a6a0bbb0df7da86e9` 时，`detection_logics/` 包含：
- `detection_logics/dns_suspicious_string.py`
- 检测原始 DNS 值（punycode 标签）和/或解码值中的已知可疑字符串。
- 评分：每发现一个独特的可疑字符串 `+1`
- 原因：`dns_suspicious_string`
- `detection_logics/vt_verdict_not_clean.py`
- 检测并权衡嵌入在文本中的非 clean VirusTotal 判定 token（例如 `vt:malicious`、`vt_verdict=suspicious`、`[vt: suspicious]`）。
- 评分：
- `malicious` → `+2`
- `suspicious`（以及其他非 clean 判定） → `+1`
- 原因：`vt_verdict_not_clean`
- `detection_logics/__init__.py`
- 注册哪些逻辑函数在 DNS 与 URL 上下文中运行，并公开：
- `apply_dns_logics()`
- `apply_url_logics()`
## 数据转换工具
[data_transform](./data_transform) 目录提供了用于清理和预处理遥测数据的命令行工具：
* **数据匿名化工具** ([data_anonymisation.py](./data_transform/data_anonymisation.py))：
- 搜索并替换敏感信息，如用户名、域登录名、计算机名和公司名称。
- 具有自动发现用户配置文件路径（`C:\Users\...`、`/home/...`）和常见主机名（例如 `DESKTOP-XXXXXXX`）的智能功能。
- 使用 `data_anonymisation.input` 中的映射进行自定义。
- 支持 `--dry-run`、`--in-place` 修改以及自定义输出后缀。
* **数据去重工具** ([data_deduplication.py](./data_transform/data_deduplication.py))：
- 执行不区分大小写的去重，并过滤掉空行。
- **数值聚合支持**：自动识别包含“occurrence”或“prevalence”（不区分大小写）的 CSV 列，并汇总重复行中的数值，将它们合并为单行。
- 支持 `--stats`、`--dry-run`、`--in-place` 修改以及自定义输出后缀。
有关使用参数和示例，请参阅 [data_transform/README.md](./data_transform/README.md)。
## 项目结构
```
├── .github/ # GitHub Actions & validation helper scripts
│ ├── scripts/
│ │ ├── create_data_transform_issues.py
│ │ ├── create_missing_catalog_issues.py
│ │ ├── create_missing_scenarios_folder_issues.py
│ │ ├── find_missing_in_catalog.py
│ │ └── find_missing_scenarios_folders.py
│ └── workflows/
│ ├── check-catalog-sync.yml
│ ├── check-data-transform.yml
│ ├── check-scenarios-folders.yml
│ ├── download-confusables.yml
│ └── virustotal-high-confidence.yml
├── data_grabber/
│ └── sentinelone-powerquery/
│ ├── sentinelone_query.py # SentinelOne PowerQuery collector
│ └── config.json # Local query configuration
├── data_transform/ # Telemetry cleaning and deduplication utilities
│ ├── data_anonymisation.py
│ ├── data_anonymisation.input.example
│ └── data_deduplication.py
├── detection_logics/ # Shared scoring and enrichment helpers (reusable rule hits)
├── install/
│ ├── bootstrap_jupyter_venv.ps1 # Central JupyterLab environment bootstrap (Windows)
│ ├── bootstrap_jupyter_venv.py # Central JupyterLab environment bootstrap (Python)
│ ├── bootstrap_scenario_venv.ps1 # Scenario-specific venv bootstrap & kernel registration (Windows)
│ ├── bootstrap_scenario_venv.sh # Scenario-specific venv bootstrap & kernel registration (Bash)
│ ├── install_dependencies.ps1 # Local dependency bootstrap
│ ├── install_dependencies.sh # Linux/macOS dependency bootstrap
│ └── requirements.txt # Shared Python dependencies
├── PEAK/ # Reference material for the PEAK framework
├── scenarios/
│ ├── catalog.csv # M-ATH use case catalog
│ ├── */README.md # Scenario documentation
│ ├── */input/ # Source telemetry or exported datasets
│ ├── */output/ # Analysis outputs and ranked findings
│ └── */*.ipynb # Scenario notebooks where implemented
├── scripts/
│ ├── add_virustotal_verdicts.py
│ ├── bootstrap_scenarios.py
│ ├── fetch_sentinelone.py
│ ├── json_to_csv.py
│ ├── run_analysis.py
│ └── update_catalog_folders.py
└── scripts/ # Utility and runner scripts
├── start_jupyterlab.ps1 # Runner to start central JupyterLab (Windows)
└── start_jupyterlab.py # Runner to start central JupyterLab (Python)
```
## 架构
| 组件 | 用途 |
|-----------|---------|
| **GitHub Actions** | 质量保证：检查目录同步、检查场景文件夹完整性、测试数据转换合规性、更新 Unicode 混淆字符并进行富化 |
| **本地执行** | 完全支持使用中央 JupyterLab 运行环境（`.jupyter_venv`）以及注册为自定义 Jupyter 内核的场景隔离环境（`.venv`） |
## 设置与本地开发
### JupyterLab 开发流程（推荐）
```
graph TD
classDef hostStyle fill:#ECEFF1,stroke:#37474F,stroke-dasharray: 5 5;
classDef venvStyle fill:#E1F5FE,stroke:#0288D1,stroke-width:2px;
classDef kernelStyle fill:#E8F5E9,stroke:#2E7D32,stroke-width:2px;
classDef libStyle fill:#FFF3E0,stroke:#F57C00,stroke-width:2px;
subgraph Host ["JupyterLab Server Host"]
JL[JupyterLab Run Environment]
end
subgraph Central ["Central Environment"]
V1[".jupyter_venv
(Server Dependencies)"]:::venvStyle
end
subgraph Kernels ["Scenario Isolated Environments"]
V2["scenarios/dga_detection/.venv
(M-ATH Kernel)"]:::kernelStyle
V3["scenarios/process_clustering/.venv
(M-ATH Kernel)"]:::kernelStyle
end
subgraph Shared ["Shared Libraries"]
DL["detection_logics
(Editable Mode)"]:::libStyle
end
JL -->|Launches with| V1
JL -->|Interacts via kernel selection| V2
JL -->|Interacts via kernel selection| V3
V2 -.->|Depends on| DL
V3 -.->|Depends on| DL
class Host hostStyle;
```
本地 notebooks 在场景隔离的虚拟环境中执行，以防止依赖冲突，并注册为 JupyterLab 的自定义内核：
1. **引导中央 JupyterLab**：
在 `.jupyter_venv` 下安装中央 JupyterLab 服务器环境：
* **Windows (PowerShell):**
.\install\bootstrap_jupyter_venv.ps1
* **Linux/macOS:**
python install/bootstrap_jupyter_venv.py
2. **引导场景虚拟环境**：
创建场景隔离的 `.venv`，以可编辑模式（`pip install -e`）安装共享的 `detection_logics` 包，并注册其自定义 Jupyter 内核（例如，`M-ATH: dga_detection`）：
* **Windows (PowerShell):**
.\install\bootstrap_scenario_venv.ps1 -ScenarioPath scenarios\dga_detection
* **Linux/macOS:**
chmod +x ./install/bootstrap_scenario_venv.sh
./install/bootstrap_scenario_venv.sh scenarios/dga_detection
3. **启动 JupyterLab**：
以无头模式启动 JupyterLab：
* **Windows (PowerShell):**
.\scripts\start_jupyterlab.ps1
* **Linux/macOS:**
python scripts/start_jupyterlab.py
从控制台输出中复制服务器 URL 和 token，在浏览器中打开它，加载场景 notebook（`.ipynb`），并从右上角的下拉菜单中选择已注册的 `M-ATH:
` 内核。
### 脚本执行（备选本地设置）
要在 JupyterLab 之外执行独立的辅助 Python 脚本：
* **Windows (PowerShell):**
.\install\install_dependencies.ps1
* **Linux/macOS:**
chmod +x ./install/install_dependencies.sh
./install/install_dependencies.sh
对于启用了 VirusTotal 的脚本，请在您的环境中设置 `VT_API_KEY`。
### GitHub Actions 集成
1. **VirusTotal 富化**：
- 在仓库设置中添加 `VT_API_KEY` secret（**Settings** → **Secrets and variables** → **Actions**）。
- 在更新高置信度发现时自动运行，或通过 workflow dispatch 手动运行。
2. **自动验证检查**：
- 自动化工作流每天凌晨 2:00（GMT）运行，以检测缺失的场景条目或文件夹，并验证脚本合规性。
### Git Pre-Commit Hooks（开发安全）
为了防止意外提交私有遥测数据或配置，请启用 pre-commit hook：
```
git config core.hooksPath .githooks
```
在 Linux/macOS 上，确保该 hook 脚本具有可执行权限：
```
chmod +x .githooks/pre-commit
```
## GitHub Workflows 与自动化
本仓库包含在 `.github/workflows/` 中配置的持续集成工作流：
* **检查目录同步** ([check-catalog-sync.yml](./.github/workflows/check-catalog-sync.yml))：
- 每天或手动运行。
- 识别未包含在 `scenarios/catalog.csv` 中的场景目录（使用 [find_missing_in_catalog.py](./.github/scripts/find_missing_in_catalog.py)），并创建 GitHub Issues（使用 [create_missing_catalog_issues.py](./.github/scripts/create_missing_catalog_issues.py)）。
* **检查场景文件夹** ([check-scenarios-folders.yml](./.github/workflows/check-scenarios-folders.yml))：
- 每天或手动运行。
- 验证 `catalog.csv` 中的所有场景是否具有包含 `input/` 和 `output/` 目录的文件夹（使用 [find_missing_scenarios_folders.py](./.github/scripts/find_missing_scenarios_folders.py)），并创建 GitHub Issues（使用 [create_missing_scenarios_folder_issues.py](./.github/scripts/create_missing_scenarios_folder_issues.py)）。
* **检查数据转换脚本** ([check-data-transform.yml](./.github/workflows/check-data-transform.yml))：
- 在向 `data_transform/` 发起 PR/推送时触发。
- 对 Python 脚本（使用 [create_data_transform_issues.py](./.github/scripts/create_data_transform_issues.py)）进行质量检查，涵盖编译、shebang 结构、顶级 docstrings 和 argparse `--dry-run` 支持，并为合规失败创建 issues。
* **下载混淆字符** ([download-confusables.yml](./.github/workflows/download-confusables.yml))：
- 在向默认分支推送或手动时自动运行。
- 动态更新 `detection_logics/resources/unicode_TR39_confusables.txt` 下的 Unicode 混淆字符数据。
* **添加 VirusTotal 判定** ([virustotal-high-confidence.yml](./.github/workflows/virustotal-high-confidence.yml))：
- 在发生更改时，自动使用 VirusTotal 判定丰富高置信度的发现。