Bandjacks

网络威胁防御世界建模系统

概述

Bandjacks 是一个全面的网络威胁情报（CTI）系统，能够：

在 12-40 秒内 从威胁报告中提取 MITRE ATT&CK 技术
构建威胁行为者、技术和防御措施的知识图谱
生成符合 STIX 2.1 的捆绑包，并带有完整的溯源追踪
集成 D3FEND 本体以提供防御建议
提供向量搜索和图分析能力
计算 共现分析 以识别技术模式
通过 LLM 响应缓存，提取速度比早期版本快 94%
包含 Next.js 前端 用于报告审核和分析可视化

📚 文档

指南	描述
快速开始	在 5 分钟内启动并运行
完整安装	完整的环境设置
CLI 使用	命令行界面指南
API 参考	REST API 文档
共现分析	分析文档
攻击流生成	流程生成指南
审核系统	人机交互审核

架构亮点

TechniqueCache

启动时加载所有 MITRE ATT&CK 技术的 内存缓存
通过 external_id（例如 T1557）进行 O(1) 查找，实现即时名称解析
缓存 1376 种技术 及其完整元数据（名称、描述、战术、平台）
一致的命名 确保审核界面始终显示人类可读的技术名称

ActorCache

所有入侵集和威胁行为者的 内存缓存
快速查找以解析行为者名称和搜索
支持别名匹配和模糊搜索

快速开始

前置条件

Python 3.11+
Neo4j 5.x（图数据库）
OpenSearch 2.x（向量存储）
Redis（可选，用于缓存）
Node.js 18+（用于前端）
LLM 访问：云 API 密钥（Gemini 或 OpenAI）或本地 OpenAI 兼容服务器

安装```bash

Clone the repository

git clone https://github.com/yourusername/bandjacks.git cd bandjacks

Install Python dependencies with uv (recommended)

uv sync

Or with pip

pip install -e .

Install frontend dependencies

cd ui && npm install && cd ..

### 环境设置

**重要：** 在启动应用程序之前，您必须配置环境变量。应用程序需要设置 `NEO4J_PASSWORD`。

在项目根目录中创建一个 `.env` 文件：

``````bash
# Copy the sample file
cp infra/env.sample .env

# Edit .env and set your actual passwords
nano .env

Required configuration in .env:```bash

Neo4j Configuration (REQUIRED)

NEO4J_URI=bolt://localhost:7687 NEO4J_USER=neo4j NEO4J_PASSWORD=your-actual-neo4j-password # MUST BE SET - no default provided

OpenSearch Configuration

OPENSEARCH_URL=http://localhost:9200 OPENSEARCH_USER=admin OPENSEARCH_PASSWORD=your-opensearch-password # Optional if security is disabled

LLM Configuration — pick ONE of the options below:

Option A: Local OpenAI-compatible API (vLLM, llama.cpp, Ollama, LocalAI, LM Studio, etc.)

LOCAL_LLM_API_BASE=http://192.168.1.100:8080/v1 # Base URL of your local server LOCAL_LLM_MODEL=mistral-nemo # Model name as the server reports it LOCAL_LLM_API_KEY=no-key # Most local servers accept any value

Option B: Cloud LLM providers

PRIMARY_LLM=gemini GOOGLE_API_KEY=your-gemini-api-key

Optional: OpenAI as fallback (or primary if PRIMARY_LLM=openai)

OPENAI_API_KEY=your-openai-api-key

ATT&CK Configuration

ATTACK_INDEX_URL=https://raw.githubusercontent.com/mitre-attack/attack-stix-data/master/index.json ATTACK_COLLECTION=enterprise-attack ATTACK_VERSION=latest

Redis (optional, for caching)

REDIS_URL=redis://localhost:6379

**注意：** 如果未设置 `NEO4J_PASSWORD`，应用程序将无法启动。详情请参阅 [Environment Variables Fix](https://github.com/blevene/bandjacks/blob/HEAD/ENV_VARIABLES_FIX.md)。

### 启动服务```bash
# Start the FastAPI backend server
uv run uvicorn bandjacks.services.api.main:app --reload --port 8000

# In another terminal, start the Next.js frontend
cd ui && npm run dev

# Access the applications
open http://localhost:8000/docs    # API documentation
open http://localhost:3000         # Frontend UI

命令行界面 (CLI)

Bandjacks 包含一个全面的用于威胁情报操作的CLI：```bash

Show all available commands

uv run python -m bandjacks.cli.main --help

> **注意：** CLI 需要设置环境变量（NEO4J_PASSWORD 等）。请从项目根目录运行，那里包含了 `.env` 文件。

### 查询命令```bash
# Search for threat intelligence
uv run python -m bandjacks.cli.main query search "ransomware encryption techniques" --top-k 10

# Explore graph relationships
uv run python -m bandjacks.cli.main query graph "attack-pattern--abc123" --depth 2

审查队列管理```bash

Show review queue

uv run python -m bandjacks.cli.main review queue --status pending --limit 20

Approve a candidate

uv run python -m bandjacks.cli.main review approve "candidate-123" --reviewer analyst-1

Reject with reason

uv run python -m bandjacks.cli.main review reject "candidate-456" --reviewer analyst-1 --reason "False positive"

### 文档提取```bash
# Extract CTI from a document
uv run python -m bandjacks.cli.main extract document ./report.pdf --confidence-threshold 80 --show-evidence

分析命令

注意： 分析命令需要Neo4j中的AttackEpisode数据才能返回结果。```bash

Show top co-occurring technique pairs

uv run python -m bandjacks.cli.main analytics top-cooccurrence --limit 25 --min-episode-size 2

Compute conditional co-occurrence P(B|A) for a technique

uv run python -m bandjacks.cli.main analytics conditional "attack-pattern--abc123" --limit 25

Analyze a specific threat actor

uv run python -m bandjacks.cli.main analytics actor "intrusion-set--xyz789" --metric npmi

Extract technique bundles

uv run python -m bandjacks.cli.main analytics bundles --min-support 3 --min-size 3 --max-size 5 --format json --output bundles.json

Global co-occurrence metrics

uv run python -m bandjacks.cli.main analytics global --min-support 2 --limit 50 --format csv --output pairs.csv

### 工作流命令```bash
# Process a directory of reports with analytics
uv run python -m bandjacks.cli.main workflow process-reports ./reports/ --workers 3 --analyze --export-dir ./results/

# Bulk export all analytics data
uv run python -m bandjacks.cli.main workflow bulk-export --export-dir ./analytics_export/

管理命令```bash

Check system health

uv run python -m bandjacks.cli.main admin health

View cache statistics

uv run python -m bandjacks.cli.main admin cache-stats

Clear cache

uv run python -m bandjacks.cli.main admin cache-clear --pattern "search:*"

Optimize database

uv run python -m bandjacks.cli.main admin optimize

## 前端用户界面

基于 Next.js 的前端提供了用于操作系统的现代界面。

### 报告管理 (`/reports`)
- **报告列表**: 查看所有已导入的报告及其状态和技术计数
- **新建报告** (`/reports/new`): 上传 PDF/TXT 文件或粘贴报告内容
- **报告详情** (`/reports/[id]`): 查看提取的技术、实体和证据
- **审核界面** (`/reports/[id]/review`): 人机协同审核工作流

### 共现分析 (`/analytics/cooccurrence`)

> **注意:** 这些页面需要 Neo4j 中存在 `AttackEpisode` 数据。请先通过提取流程处理报告，或使用 `POST /v1/flows/build` 从入侵集数据生成事件。

- **中心页面**: 概览页面，显示事件/技术/行为者计数
- **热门配对** (`/pairs`): 共现技术配对及其 NPMI/提升度指标
- **条件概率** (`/conditional`): P(B|A) 条件概率
- **捆绑** (`/bundles`): 频繁共现的技术捆绑集
- **行为者** (`/actors`): 特定行为者的技术模式
- **桥接** (`/bridging`): 跨多个行为者使用的技术

### 系统健康 (`/health`)
- 所有组件（Neo4j、OpenSearch、Redis）的实时健康状态
- 缓存统计信息和内存使用情况
- 与 Kubernetes 兼容的健康端点

### 启动前端```bash
cd ui
npm run dev     # Development mode with hot reload
npm run build   # Production build
npm run start   # Start production server

# Ensure backend is running
# API_URL defaults to http://localhost:8000/v1

使用指南

1. 加载 MITRE ATT&CK 数据

首先，将 MITRE ATT&CK 框架加载到您的知识图谱中：```bash

Load the latest enterprise ATT&CK release

curl -X POST "http://localhost:8000/v1/stix/load/attack"
-H "Content-Type: application/json"
-d '{ "collection": "enterprise-attack", "version": "latest", "adm_strict": false }'

### 2. 从报告中提取技术

从威胁情报报告中提取MITRE ATT&CK技术：```python
import httpx
import time

# For small reports (<5KB) - synchronous processing
response = httpx.post(
    "http://localhost:8000/v1/reports/ingest",
    json={
        "content": "APT29 used spearphishing emails with malicious attachments...",
        "title": "APT29 Campaign Analysis",
        "config": {
            "use_optimized_extractor": True,
            "span_score_threshold": 0.7,
            "top_k": 5
        }
    }
)

result = response.json()
print(f"Extracted {len(result['extraction']['techniques'])} techniques")

# For large reports (>5KB) - asynchronous processing
response = httpx.post(
    "http://localhost:8000/v1/reports/ingest_async",
    json={
        "content": large_report_text,
        "title": "Large Report Analysis"
    }
)

job_id = response.json()["job_id"]

# Check job status
status = httpx.get(f"http://localhost:8000/v1/reports/jobs/{job_id}/status")
while status.json()["status"] == "processing":
    time.sleep(2)
    status = httpx.get(f"http://localhost:8000/v1/reports/jobs/{job_id}/status")

# Get results from completed job
result = status.json()["result"]
print(f"Extracted {result['techniques_count']} techniques in {result['elapsed_time']} seconds")

3. 直接使用 Python

用于无需 API 的程序化访问：```python from bandjacks.llm.extraction_pipeline import run_extraction_pipeline

Configure extraction

config = { "use_optimized_extractor": True, # Use optimized pipeline "span_score_threshold": 0.7, # Minimum span confidence "max_spans": 20, "top_k": 5, "chunk_size": 2000, # For large documents "max_chunks": 100 }

Run extraction pipeline

result = run_extraction_pipeline( report_text, config, source_id="report_123", neo4j_config=neo4j_config )

Access results

techniques = result["techniques"] # Dict of technique_id -> details bundle = result.get("bundle") # STIX 2.1 bundle if configured entities = result.get("entities") # Extracted entities

Example: Print extracted techniques

for tech_id, info in techniques.items(): print(f"{tech_id}: {info['name']}") print(f" Confidence: {info['confidence']}%") print(f" Evidence: {info['evidence']}")

## 提取管道架构

Bandjacks 提取管道采用多智能体架构来提取结构化威胁情报：

### 管道组件

提取管道按顺序使用 9 个专用智能体：

#### 1. **EntityExtractionAgent** - 实体识别
- 提取威胁行为者、恶意软件、工具和活动
- 首先运行，为技术提取提供上下文
- 使用少样本提示与 JSON schema 验证
- 处理分块文档，采用渐进式窗口提取

#### 2. **SpanFinderAgent** - 行为文本检测
- 使用 14 个特定于策略的正则表达式模式检测包含威胁行为的文本跨度
- 识别显式技术 ID（T1566.001）和行为模式
- 通过关键词索引提升对跨度进行置信度评分
- 无 LLM 调用——纯模式匹配以提高速度

#### 3. **BatchRetrieverAgent** - 候选检索
- 使用 OpenSearch KNN 向量搜索为每个跨度查找候选技术
- 在编码前对重复的跨度文本进行去重，避免冗余嵌入
- 返回每个跨度的 top-k 候选及其相似度评分

#### 4. **Pre-filter** - 跨度精简
- 每个候选技术的跨度数量限制为 `max_spans_per_technique`（默认 2）
- 保留每个候选的最高得分跨度，以保持证据质量
- 减少映射器 LLM 调用约 46%，且技术损失最小

#### 5. **DiscoveryAgent** - LLM 发现（条件触发）
- 当检索器置信度低（平均 <0.7）时触发
- 使用 LLM 发现向量搜索遗漏的技术
- 对所有低置信度跨度进行单次批量调用

#### 6. **BatchMapperAgent** - 技术映射（LLM）
- 批量处理跨度，每批最多 10 个（`MAX_MAPPER_BATCH_SIZE`，2026 年 5 月从默认 25 降低以限制云端 LLM 截断）
- 提取每个跨度的所有相关技术及其置信度评分
- 使用 JSON schema 验证确保结构化输出

#### 7. **EvidenceVerifierAgent** - 证据验证
- 基于模式的引用文本和行号验证
- 在 40-100 分范围内对证据质量进行评分
- 无 LLM 调用——正则表达式与文本匹配

#### 8. **ConsolidatorAgent** - 证据合并
- 合并跨多个跨度发现的重复技术
- 使用 Jaccard 相似度（>85% 阈值）聚合证据
- 生成最终技术列表及合并后的置信度评分

#### 9. **AttackFlowSynthesizer** - 序列生成（LLM）
- 分析时间标记（"first"、"then"、"after"）
- 从叙述中推断因果关系
- 创建带有概率边的 STIX Attack Flow 对象
- 当序列不明确时回退到共现建模

### 性能优化

- **智能分块**：文档按 2KB 块分割并重叠
- **批量处理**：映射器每次 LLM 调用最多处理 25 个跨度
- **并行处理**：块在工作线程中并发处理
- **响应缓存**：缓存 LLM 响应以避免重复调用
- **提前终止**：高置信度提取跳过验证
- **TechniqueCache**：所有 ATT&CK 技术在启动时加载，实现 O(1) 查询
- **Pre-filter**：在 LLM 映射器前限制每个候选技术的跨度数量（减少 46% 调用）
- **批量嵌入**：技术嵌入批量生成（速度提高 2-5 倍）
- **连接池**：跨请求共享 Neo4j/OpenSearch 连接
- **UNWIND 批处理**：通过 UNWIND 批量写入 Neo4j（30-40 次查询 → 6-7 次）
- **模型预热**：嵌入模型在启动时加载，避免冷启动延迟

### 处理时间

| 文档大小 | 处理时间 | 提取的技术数量 |
|--------------|-----------------|---------------------|
| 小型（<5KB） | 10-20 秒 | 5-10 种技术 |
| 中型（5-15KB） | 20-40 秒 | 10-15 种技术 |
| 大型（>15KB） | 30-60 秒 | 15-25 种技术 |

## 共现分析

Bandjacks 提供用于理解技术关系的分析功能。

> **注意：** 分析需要 `AttackEpisode` 和 `AttackAction` 数据存在于 Neo4j 中。这些数据在以下情况创建：
> - 报告通过提取管道处理
> - 攻击流通过 `/v1/flows/build` 构建
> - 包含攻击剧情的 STIX 数据包被摄入
>
> 如果不存在剧情，分析将返回空结果。

### 全局共现

计算哪些技术在所有攻击剧情中频繁同时出现：```python
# Via API
response = httpx.post(
    "http://localhost:8000/v1/analytics/cooccurrence/global",
    json={"min_support": 2, "min_episodes_per_pair": 2, "limit": 50}
)

for pair in response.json()["pairs"]:
    print(f"{pair['name_a']} + {pair['name_b']}: NPMI={pair['npmi']:.3f}")

条件概率

计算 P(B|A) - 给定使用了技术 A，技术 B 的概率是多少：```python response = httpx.get( "http://localhost:8000/v1/analytics/cooccurrence/conditional", params={"technique_id": "attack-pattern--abc123", "limit": 25} )

### 技术束

识别频繁共现的技术束（3-5个技术）：```python
response = httpx.post(
    "http://localhost:8000/v1/analytics/cooccurrence/bundles",
    json={"min_support": 3, "min_size": 3, "max_size": 5}
)

针对特定行为者的分析

分析特定威胁行为者的技术模式：```python response = httpx.post( "http://localhost:8000/v1/analytics/cooccurrence/actor", json={"intrusion_set_id": "intrusion-set--xyz789", "min_support": 1} )

## Human-in-the-Loop Review System

Bandjacks includes a comprehensive review system for validating extracted intelligence:

### Unified Review Interface

The review system presents all extracted items in a single interface:```typescript
// Review workflow
1. Upload/ingest report → Extraction pipeline runs
2. Navigate to /reports/{id}/review
3. Review extracted items across three tabs:
   - Entities (threat actors, malware, tools)
   - Techniques (ATT&CK mappings with evidence)
   - Attack Flow (sequenced steps)
4. Take actions on each item:
   - Approve: Accept as correct
   - Reject: Mark as incorrect
   - Edit: Modify details (name, confidence, etc.)
5. Submit all decisions atomically

审查功能

证据链接：直接链接到带有行号的源文本
置信度调整：根据分析师知识修改置信度分数
批量操作：选择多个项目进行批量批准/拒绝
键盘快捷键：A（批准），R（拒绝），E（编辑），空格（下一个）
进度跟踪：审查完成的视觉指示器
过滤：按类型、置信度水平或状态过滤

API 集成```python

Submit review decisions

response = httpx.post( f"http://localhost:8000/v1/reports/{report_id}/unified-review", json={ "decisions": [ { "item_id": "technique-0", "action": "approve", "confidence_adjustment": 5, "notes": "Confirmed via external CTI" }, { "item_id": "entity-malware-1", "action": "edit", "edited_value": { "name": "Corrected Malware Name", "confidence": 95 } } ], "global_notes": "Review completed by analyst-1" } )

Review creates:

- Approved entities as Neo4j nodes

- Technique-to-report relationships

- Audit trail of decisions

### 4. 搜索技术

使用自然语言搜索 ATT&CK 技术：```python
# Vector search for similar techniques
response = httpx.post(
    "http://localhost:8000/v1/search/ttx",
    json={
        "query": "ransomware that encrypts files and demands payment",
        "top_k": 5
    }
)

techniques = response.json()["results"]
for tech in techniques:
    print(f"{tech['external_id']}: {tech['name']} (score: {tech['score']:.2f})")

5. 图查询

查询知识图谱中的关系：```python

Get all techniques used by a specific group

response = httpx.get( "http://localhost:8000/v1/graph/group/G0016/techniques" )

Get defensive techniques for an attack

response = httpx.get( "http://localhost:8000/v1/defense/technique/T1566.001" )

### 6. 生成攻击流模型

创建共现模型，展示威胁行为者如何协同使用技术：```python
# Generate flow for a specific intrusion set (e.g., APT29)
response = httpx.post(
    "http://localhost:8000/v1/flows/build",
    json={
        "intrusion_set_id": "intrusion-set--899ce53f-13a0-479b-a0e4-67d46e241542"
    }
)

flow = response.json()
print(f"Generated flow '{flow['name']}' with {len(flow['steps'])} techniques")
print(f"Co-occurrence edges: {len(flow['edges'])}")

批量生成：为所有威胁行为者及其技术生成流程：```bash

Run the bulk generation script

uv run python scripts/build_intrusion_flows_simple.py

Monitor progress - creates flows for 165+ intrusion sets

Handles rate limiting automatically

Skips existing flows to avoid duplicates

AttackFlow 模型使用 **共现** 而非顺序排序，因为入侵集没有固有的顺序信息。技术通过以下方式连接：
- **战术内边**：同一杀伤链战术内的技术之间
- **跨战术边**：相邻战术之间的技术之间
- **中心辐射模式**：用于大型技术集以避免边爆炸

详细用法请参见 [AttackFlow 生成指南](https://github.com/blevene/bandjacks/blob/HEAD/docs/ATTACKFLOW_GENERATION.md)。

## 支持的输入格式

提取管道支持多种输入格式：

- **纯文本** - 直接文本内容
- **Markdown** - 格式化的 Markdown 文档
- **PDF** - 通过 pdfplumber 提取
- **HTML** - 通过 BeautifulSoup 解析
- **JSON** - 结构化数据提取

### 从纯文本提取```python
# Direct text extraction
plaintext_report = """
The threat actors used spearphishing emails with malicious attachments.
After gaining access, they deployed Mimikatz to harvest credentials and
used RDP for lateral movement across the network.
"""

result = asyncio.run(run_agentic_v2_async(plaintext_report, {
    "cache_llm_responses": True,
    "single_pass_threshold": 500
}))

从 Markdown 中提取```python

Markdown document extraction

markdown_report = """

APT Campaign Analysis

Attack Methods

Initial Access: Spearphishing with malicious Office documents
Execution: PowerShell scripts and scheduled tasks
Persistence: Registry modifications and service installation

Tools Used

Tool	Purpose
Mimikatz	Credential dumping
PsExec	Remote execution
Cobalt Strike	C2 communications
"""

result = run_extraction_pipeline(markdown_report, { "use_optimized_extractor": True, "span_score_threshold": 0.7 }, source_id="markdown_report")

### 从 PDF 中提取```python
import pdfplumber
from bandjacks.llm.extraction_pipeline import run_extraction_pipeline

# Read PDF with pdfplumber (recommended)
with pdfplumber.open("threat_report.pdf") as pdf:
    text = ""
    for page in pdf.pages:
        page_text = page.extract_text()
        if page_text:
            text += page_text + "\n"

# Extract techniques using extraction pipeline
result = run_extraction_pipeline(text, {
    "use_optimized_extractor": True,
    "span_score_threshold": 0.7,
    "chunk_size": 2000
}, source_id="threat_report")

print(f"Found {len(result['techniques'])} techniques")

批量处理报告```python

from pathlib import Path import json

reports_dir = Path("./reports") results = []

for pdf_file in reports_dir.glob("*.pdf"): # Extract text and techniques # ... (see above)

results.append({
    "file": pdf_file.name,
    "techniques": list(result["techniques"].keys()),
    "count": len(result["techniques"])
})

Save summary

with open("extraction_summary.json", "w") as f: json.dump(results, f, indent=2)

### 构建攻击流```python
# Generate attack flow from extracted techniques
response = httpx.post(
    "http://localhost:8000/v1/flows/build",
    json={
        "source_id": "report-123",
        "technique_ids": ["T1566.001", "T1059.001", "T1003.001"]
    }
)

flow = response.json()
print(f"Generated flow with {len(flow['steps'])} steps")

测试

运行测试套件以验证您的安装：```bash

Run all tests

uv run pytest

Test extraction pipeline

python tests/test_optimized_extraction.py

Test graph integration

python tests/test_graph_upsert.py

Test STIX validation

python tests/test_bundle_validation.py

Run frontend tests

cd ui && npm test

## API 端点

### 核心端点

- `POST /v1/stix/load/attack` - 加载 MITRE ATT&CK 数据
- `POST /v1/reports/ingest` - 同步报告导入（<5KB）
- `POST /v1/reports/ingest_async` - 异步报告导入（>5KB）
- `POST /v1/reports/ingest/upload` - 上传 PDF/TXT 文件
- `GET /v1/reports/jobs/{id}/status` - 检查任务状态
- `POST /v1/reports/{id}/unified-review` - 提交审阅决定
- `POST /v1/search/ttx` - 搜索技术
- `GET /v1/graph/technique/{id}` - 获取技术详情

### 攻击流

- `POST /v1/flows/build` - 生成 AttackFlow 共现模型
- `GET /v1/flows/{flow_id}` - 检索特定 AttackFlow 详情
- `POST /v1/flows/search` - 搜索相似攻击流
- `GET /v1/flows/dump` - 批量导出带有分页和过滤的流

### 分析

- `GET /v1/analytics/cooccurrence/global` - 全局共现指标
- `GET /v1/analytics/cooccurrence/conditional` - 条件概率
- `GET /v1/analytics/cooccurrence/bundles` - 技术包
- `GET /v1/analytics/cooccurrence/actor` - 特定于行为者的模式
- `GET /v1/coverage/gaps` - 技术覆盖缺口

### 防御与检测

- `GET /v1/defense/technique/{id}` - 获取防御建议
- `GET /v1/detections/technique/{id}` - 检测策略
- `POST /v1/sigma/validate` - 验证 Sigma 规则

### 监控

- `GET /health` - 基本健康检查
- `GET /health/live` - Kubernetes 存活探测
- `GET /health/ready` - Kubernetes 就绪探测
- `GET /health/components/{component}` - 单个组件健康状态
- `GET /v1/costs/stats` - LLM 成本跟踪（按模型每日汇总）
- `GET /v1/cache/stats` - 获取 LLM 缓存统计
- `POST /v1/cache/clear` - 清除 LLM 缓存
- `GET /v1/compliance/report` - 合规指标
- `GET /v1/drift/status` - 漂移检测状态
- `GET /v1/ml-metrics/performance` - ML 模型指标

### 行为者与来源

- `GET /v1/actors` - 列出威胁行为者
- `GET /v1/actors/{id}` - 获取行为者详情
- `GET /v1/provenance/{object_id}` - 对象来源
- `GET /v1/provenance/{object_id}/lineage` - 完整溯源链
- `GET /v1/provenance/{object_id}/evidence` - 证据片段

### 仅 API 功能（无 UI/CLI）

这些端点功能完整，但仅通过 REST API 访问（无前端页面或 CLI 命令）：

#### 攻击路径模拟
- `POST /v1/simulation/paths` - 从起始技术/组织模拟攻击路径
- `POST /v1/simulation/predict` - 根据当前状态预测下一步可能的技术
- `POST /v1/simulation/whatif` - 防御场景的假设分析
- `POST /v1/simulation/scenario` - 从一组组织、软件、技术进行模拟
- `GET /v1/simulation/statistics/{technique_id}` - 技术使用统计
- `GET /v1/simulation/groups/{group_id}/patterns` - 组织攻击模式
- `POST /v1/simulation/compare` - 比较多个攻击路径

#### MDP 策略与推出
- `POST /v1/simulate/rollout` - PTG 推出模拟
- `POST /v1/simulate/mdp` - 计算 MDP 最优防御策略
- `GET /v1/simulate/models` - 列出可用的 PTG 模型

#### 漂移检测与监控
- `GET /v1/drift/status` - 所有指标的当前漂移状态
- `POST /v1/drift/analyze` - 使用自定义阈值运行漂移分析
- `GET /v1/drift/alerts` - 获取活跃漂移告警
- `POST /v1/drift/alerts/{alert_id}/acknowledge` - 确认告警
- `GET /v1/drift/metrics/{metric_name}` - 获取特定漂移指标

#### ML 指标追踪
- `POST /v1/ml-metrics/prediction` - 记录模型预测以进行追踪
- `POST /v1/ml-metrics/review` - 记录审阅决策指标
- `POST /v1/ml-metrics/coverage-gap` - 记录覆盖缺口
- `GET /v1/ml-metrics/performance` - 获取模型性能指标
- `GET /v1/ml-metrics/dashboard` - 导出仪表盘指标

#### 通知
- `GET /v1/notifications/history` - 获取通知历史
- `POST /v1/notifications/clear-history` - 清除通知历史
- `GET /v1/notifications/config` - 获取通知配置
- `POST /v1/notifications/test` - 发送测试通知

#### 向量更新管理
- `GET /v1/vectors/status` - 向量更新系统状态
- `GET /v1/vectors/metrics` - 详细向量更新指标
- `POST /v1/vectors/update` - 手动触发向量更新
- `POST /v1/vectors/process-batch` - 强制批处理
- `DELETE /v1/vectors/queue` - 清除待处理更新队列
- `GET /v1/vectors/health` - 向量系统健康检查

#### 实体忽略列表
- `GET /v1/ignorelist` - 获取当前忽略列表状态
- `POST /v1/ignorelist/add` - 将实体添加到忽略列表
- `DELETE /v1/ignorelist/remove` - 从忽略列表中移除实体
- `POST /v1/ignorelist/reload` - 从磁盘重新加载忽略列表

#### 候选模式审查
- `GET /v1/review/candidates` - 列出候选攻击模式
- `POST /v1/review/candidates` - 创建候选模式
- `GET /v1/review/candidates/{id}` - 获取候选详情
- `POST /v1/review/candidates/{id}/approve` - 批准候选
- `POST /v1/review/candidates/{id}/reject` - 拒绝候选
- `GET /v1/review/candidates/{id}/similar` - 查找相似模式
- `GET /v1/review/candidates/stats/summary` - 候选统计

### 完整 API 文档

访问完整 API 文档：
- Swagger UI: http://localhost:8000/docs
- ReDoc: http://localhost:8000/redoc
- OpenAPI JSON: http://localhost:8000/openapi.json

## 架构

### 项目结构```
bandjacks/
├── bandjacks/
│   ├── analysis/         # Graph analysis & interdiction
│   │   ├── graph_analyzer.py
│   │   └── interdiction.py
│   ├── analytics/        # Co-occurrence & clustering
│   │   ├── clustering.py
│   │   ├── cooccurrence.py
│   │   └── detection_bundles.py
│   ├── cli/              # Command-line interface
│   │   ├── main.py       # CLI entry point
│   │   ├── batch_extract.py
│   │   ├── formatters.py
│   │   └── workflows.py
│   ├── config/           # Configuration files
│   │   └── entity_ignorelist.yaml
│   ├── core/             # Core utilities
│   │   ├── cache.py      # Redis caching
│   │   ├── connection_pool.py
│   │   └── query_optimizer.py
│   ├── llm/              # Extraction pipeline
│   │   ├── extraction_pipeline.py
│   │   ├── agents_v2.py  # Core extraction agents
│   │   ├── chunked_extractor.py
│   │   ├── optimized_chunked_extractor.py
│   │   ├── entity_extractor.py
│   │   ├── flow_builder.py
│   │   ├── cache.py      # LLM response caching
│   │   └── experimental/ # Experimental features
│   ├── loaders/          # Data loading & indexing
│   │   ├── attack_catalog.py
│   │   ├── attack_upsert.py
│   │   ├── opensearch_index.py
│   │   ├── hybrid_search.py
│   │   └── sigma_loader.py
│   ├── monitoring/       # Metrics & monitoring
│   │   ├── compliance_metrics.py
│   │   ├── defense_metrics.py
│   │   ├── drift_detector.py
│   │   └── ml_metrics.py
│   ├── services/         # API & services
│   │   ├── api/          # FastAPI application
│   │   │   ├── main.py
│   │   │   ├── routes/   # API route handlers
│   │   │   └── middleware/
│   │   ├── technique_cache.py
│   │   └── actor_cache.py
│   ├── simulation/       # Attack simulation
│   │   ├── attack_simulator.py
│   │   ├── mdp_solver.py
│   │   └── ptg_rollout.py
│   └── store/            # Data stores
│       ├── report_store.py
│       ├── candidate_store.py
│       └── review_store.py
├── ui/                   # Next.js frontend
│   ├── app/              # App Router pages
│   │   ├── reports/      # Report management
│   │   ├── analytics/    # Analytics dashboards
│   │   └── health/       # Health monitoring
│   ├── components/       # React components
│   └── hooks/            # Custom React hooks
├── tests/                # Test suite
├── samples/              # Sample reports
├── scripts/              # Utility scripts
└── docs/                 # Documentation

组件

提取管道 (bandjacks/llm/)
- extraction_pipeline.py - 主提取编排器
- chunked_extractor.py - 标准分块处理
- optimized_chunked_extractor.py - 高级优化处理
- agents_v2.py - 核心提取代理 (SpanFinder, Mapper, Consolidator)
- entity_extractor.py - 实体识别代理
- flow_builder.py - 攻击流生成
- memory.py - 共享工作内存
- cache.py - LLM 响应缓存
数据层 (bandjacks/loaders/)
- Neo4j 属性图用于关系
- OpenSearch 用于向量嵌入
- STIX 2.1 数据模型
API 层 (bandjacks/services/api/)
- FastAPI REST 端点
- WebSocket 支持实时更新

性能

提取速度：每份报告 12-40 秒（比 v1 快 94%）
小型文档：单次提取 4-8 秒
缓存命中率：重复提取加速 87.5%
搜索：向量相似性搜索 <300ms
图形查询：大多数遍历 <100ms

配置

模型选择

系统支持云端 LLM 和任何本地 OpenAI 兼容 API：```bash

In your .env file

--- Option A: Local inference (highest priority when set) ---

Works with vLLM, llama.cpp (server), Ollama, LocalAI, LM Studio,

text-generation-webui, or any server that exposes an /v1/chat/completions endpoint.

LOCAL_LLM_API_BASE=http://192.168.1.100:8080/v1 LOCAL_LLM_MODEL=mistral-nemo LOCAL_LLM_API_KEY=no-key # optional — most local servers don't require a key

--- Option B: Cloud providers ---

PRIMARY_LLM=gemini # "gemini" (default) or "openai" GOOGLE_API_KEY=your-key # Gemini OPENAI_API_KEY=your-key # OpenAI (used as fallback when Gemini is primary)

**提供者优先级：** 本地 API > Gemini > OpenAI > LiteLLM 代理。
当本地服务器配置后，云提供商会自动作为后备选项添加。

#### 常见本地服务器示例

| 服务器 | `LOCAL_LLM_API_BASE` | `LOCAL_LLM_MODEL` |
|--------|---------------------|-------------------|
| vLLM | `http://host:8000/v1` | `mistralai/Mistral-Nemo-Instruct-2407` |
| llama.cpp | `http://host:8080/v1` | `mistral-nemo` |
| Ollama | `http://host:11434/v1` | `mistral-nemo` |
| LM Studio | `http://host:1234/v1` | `mistral-nemo` |
| LocalAI | `http://host:8080/v1` | `mistral-nemo` |

### 提取配置

该系统使用单一的高性能异步管道，并带有可配置选项：```python
{
    "cache_llm_responses": True,         # Enable LLM caching (default: True)
    "single_pass_threshold": 500,        # Max words for single-pass (default: 500)
    "early_termination_confidence": 90,  # Skip verification above this (default: 90)
    "disable_discovery": False,          # Disable LLM discovery agent
    "max_spans": 20,                     # Maximum spans to process
    "span_score_threshold": 0.7,         # Minimum span quality
    "top_k": 5,                          # Candidates per span

    # Cost optimization options
    "max_spans_per_technique": 2,        # Pre-filter: max spans per candidate technique (0=disable, default=2)
    "enable_span_dedup": False,          # Text-based span dedup before mapping (default=False)
}

成本优化

提取管道通过 litellm.completion_cost() 跟踪 LLM 成本，提供每份报告指标和一个每日聚合端点。

成本控制：

监控：```bash

Daily cost aggregate by model

curl http://localhost:8000/v1/costs/stats

Per-report cost in extraction metrics

curl http://localhost:8000/v1/reports/{id} # -> extraction.metrics.cost_usd

### 置信度阈值

控制提取质量：```python
{
    "confidence_threshold": 50.0,  # Minimum confidence (0-100)
    "auto_ingest": True            # Auto-add high-confidence results
}

健康监控

该API为运维监控和Kubernetes部署提供全面的健康监控端点：

健康端点```bash

Basic health check (always returns 200 if API is running)

curl http://localhost:8000/health

Kubernetes liveness probe (process alive check)

curl http://localhost:8000/health/live

Kubernetes readiness probe (full dependency checks)

curl http://localhost:8000/health/ready

Individual component health

curl http://localhost:8000/health/components/neo4j curl http://localhost:8000/health/components/opensearch curl http://localhost:8000/health/components/redis curl http://localhost:8000/health/components/caches curl http://localhost:8000/health/components/system

### 健康响应示例```json
{
  "status": "healthy",
  "timestamp": "2025-01-28T17:43:30.184036Z",
  "version": "1.0.0",
  "components": {
    "neo4j": {
      "status": "healthy",
      "latency_ms": 5
    },
    "opensearch": {
      "status": "degraded",
      "cluster_status": "yellow",
      "indices": {
        "attack_nodes": false,
        "bandjacks_reports": true
      }
    },
    "redis": {
      "status": "healthy",
      "latency_ms": 2,
      "memory_mb": 1.69
    },
    "caches": {
      "status": "healthy",
      "technique_cache": {
        "count": 993,
        "loaded": true
      },
      "actor_cache": {
        "count": 145,
        "loaded": true
      }
    },
    "system": {
      "status": "healthy",
      "memory": {
        "available_gb": 8.84,
        "percent_used": 72.4
      },
      "disk": {
        "available_gb": 353.11,
        "percent_used": 2.9
      },
      "cpu": {
        "percent_used": 7.7
      }
    }
  }
}

状态级别

健康：组件完全正常运行
降级：部分功能正常（例如，缺失某些索引但仍可运行）
不健康：组件故障或不可达

Kubernetes 集成

对于 Kubernetes 部署，按如下方式配置探针：```yaml livenessProbe: httpGet: path: /health/live port: 8000 initialDelaySeconds: 30 periodSeconds: 10

readinessProbe: httpGet: path: /health/ready port: 8000 initialDelaySeconds: 45 periodSeconds: 5

## 性能优化

### 缓存

系统包含自动的LLM响应缓存，以提高性能：```python
# Check cache statistics
response = httpx.get("http://localhost:8000/v1/cache/stats")
stats = response.json()
print(f"Cache hit rate: {stats['hit_rate']}")

# Clear cache if needed
httpx.post("http://localhost:8000/v1/cache/clear")

性能配置

根据您的需求选择一个配置：```python

Fast extraction (4-15 seconds)

fast_config = { "single_pass_threshold": 1000, "max_spans": 5, "skip_verification": True, "top_k": 3 }

Balanced (default, 12-40 seconds)

balanced_config = { "single_pass_threshold": 500, "max_spans": 10, "early_termination_confidence": 90, "top_k": 5 }

High quality (40-120 seconds)

quality_config = { "single_pass_threshold": 200, "max_spans": 20, "disable_discovery": False, "min_quotes": 3, "top_k": 10 }

## 安全性

### 输入验证

- **Cypher 注入防护**：所有图查询端点均对用户提供的 `relationship_types` 参数进行验证，对照已知关系类型（USES、MITIGATES、HAS_TACTIC 等）的白名单以及严格的正则表达式模式（`^[A-Z][A-Z0-9_]*$`）。无效输入在查询构造前返回 400 错误。
- **JSON Schema 验证**：对 LLM 响应进行 JSON Schema 验证，以防止格式错误的数据进入管道。
- **ADM 验证**：所有 STIX 内容在摄入前必须通过 ATT&CK 数据模型验证。

### 身份验证与授权

- **JWT 身份验证**：用于 API 身份验证的可选中间件（`JWTAuthMiddleware`）
- **速率限制**：按端点进行速率限制，具有可配置阈值
- **CORS**：可配置的跨域资源共享

## 高级功能

### 溯源追踪

每个提取的实体都包含完整的溯源信息：```python
# Get provenance for an object
response = httpx.get(
    "http://localhost:8000/v1/provenance/attack-pattern--abc123"
)

主动学习

系统包含一个用于改进提取的审查队列：```python

Get next item for review

response = httpx.get("http://localhost:8000/v1/review_queue/next")

Submit feedback

response = httpx.post( "http://localhost:8000/v1/feedback/extraction", json={ "extraction_id": "ext-123", "correct": True, "corrections": [] } )

### 覆盖分析

分析您的威胁情报覆盖范围：```python
# Get coverage analysis
response = httpx.get("http://localhost:8000/v1/analytics/coverage")
coverage = response.json()

print(f"Summary: {coverage['summary']}")
for tactic in coverage['tactics']:
    print(f"  {tactic['tactic']}: {tactic['coverage_percentage']}%")

注意： 平台覆盖率（_analyze_platforms_coverage）目前返回占位数据。战术和组覆盖率使用真实的Neo4j查询。

攻击模拟（实验性）

模拟模块提供基于MDP的攻击路径预测：```python

Note: This feature is experimental and may require additional setup

from bandjacks.simulation.attack_simulator import AttackSimulator from bandjacks.simulation.mdp_solver import MDPSolver

See bandjacks/simulation/ for implementation details

## 功能状态

此部分提供了各项功能实施状态的透明度信息：

### 完全功能 ✅
- **报告提取管道** - 基于 LLM 的技术提取端到端工作
- **MITRE ATT&CK 加载** - 将企业/移动/ICS ATT&CK 数据加载到 Neo4j
- **向量搜索** - 基于 OpenSearch 的技术语义搜索
- **审核系统** - 通过 API 和 UI 实现的人工在环审核工作流
- **健康监控** - 组件健康检查与 Kubernetes 探针
- **CLI 查询/管理命令** - 搜索、图遍历、缓存管理
- **攻击流生成** - 基于共现的入侵集流构建
- **攻击模拟** - 基于 MDP 的路径模拟，通过 `/simulation/*` 和 `/simulate/*`
- **覆盖率报告** - 面向高管、技术、战术、运营视图的 JSON 报告

### 具有数据依赖性的功能 ⚠️
- **共现分析** - 需要报告处理生成的 `AttackEpisode` 节点
- **行为者分析** - 需要归属于入侵集的事件记录
- **技术包** - 需要足够的事件数据进行模式挖掘
- **CLI 分析命令** - 能够运行，但如果没有事件数据则返回空结果

### 仅 API（无 UI/CLI）🔌
以下功能已完全实现，仅通过 REST API 访问：
- **攻击路径模拟** - `/simulation/*` 路由，用于路径预测和假设分析
- **MDP 策略求解器** - `/simulate/mdp`，用于最优防御策略计算
- **漂移检测** - `/drift/*` 路由，用于监控数据质量漂移
- **ML 指标** - `/ml-metrics/*`，用于跟踪模型随时间变化的性能
- **向量管理** - `/vectors/*`，用于管理向量嵌入
- **实体忽略列表** - `/ignorelist/*`，用于过滤误报实体
- **候选模式** - `/review/candidates/*`，用于新颖技术候选
- **通知** - `/notifications/*`，用于告警配置和历史记录
- **溯源** - `/provenance/*`，用于提取血缘追踪
- **合规性** - `/compliance/*`，用于合规性指标报告

### 实验性（位于 `llm/experimental/`）🧪
- **PTG（概率威胁图）** - 核心逻辑已实现，测试有限
- **法官集成** - 基于 LLM 的序列验证
- **攻击流模拟器** - 基于流的模拟引擎
- **序列提取器** - 从流中提取序列

### 已移除/清理 🗑️
以下存根功能已从 API 中移除：
- ~~平台覆盖分析~~ - 之前返回硬编码的存根数据
- ~~趋势分析~~ - 之前返回随机合成数据
- ~~CSV/PDF 报告导出~~ - 之前返回 501；现仅支持 JSON
- ~~Gemini 序列推断~~ - 之前为 501 存根；请改用 `/sequence/propose`

### 连接矩阵

| 功能领域 | 前端 UI | CLI | REST API |
|--------------|-------------|-----|----------|
| 报告管理 | ✅ | ✅ | ✅ |
| 审核工作流 | ✅ | ✅ | ✅ |
| 搜索 (TTX) | ✅ | ✅ | ✅ |
| 共现分析 | ✅ | ✅ | ✅ |
| 覆盖率分析 | ✅ | - | ✅ |
| 健康监控 | ✅ | - | ✅ |
| 检测/Sigma | ✅ | - | ✅ |
| 攻击流 | ✅ | - | ✅ |
| 防御覆盖 | ✅ | - | ✅ |
| 序列/PTG | ✅ | - | ✅ |
| 行为者 | ✅ | - | ✅ |
| 攻击模拟 | - | - | ✅ |
| 漂移检测 | - | - | ✅ |
| ML 指标 | - | - | ✅ |
| 向量管理 | - | - | ✅ |
| 实体忽略列表 | - | - | ✅ |
| 候选模式 | - | - | ✅ |
| 通知 | - | - | ✅ |
| 溯源 | - | - | ✅ |
| 合规性 | - | - | ✅ |

### 前端页面
| 页面 | 状态 | 备注 |
|------|--------|-------|
| `/reports` | ✅ 工作中 | 列表、创建、查看报告 |
| `/reports/[id]/review` | ✅ 工作中 | 完整审核工作流 |
| `/analytics/cooccurrence` | ⚠️ 依赖数据 | 如有事件数据则显示 KPI |
| `/analytics/cooccurrence/pairs` | ⚠️ 依赖数据 | 调用真实 API |
| `/analytics/cooccurrence/bundles` | ⚠️ 依赖数据 | 调用真实 API |
| `/analytics/cooccurrence/actors` | ⚠️ 依赖数据 | 调用真实 API |
| `/health` | ✅ 工作中 | 实时健康状态 |

## 故障排查

### 常见问题

1. **OpenSearch 连接失败**
   - 确保 OpenSearch 正在运行：`curl http://localhost:9200`
   - 检查索引是否存在：`curl http://localhost:9200/bandjacks_attack_nodes-v1`

2. **Neo4j 连接失败**
   - 验证 Neo4j 是否正在运行：`neo4j status`
   - 检查 `.env` 文件中是否设置了 `NEO4J_PASSWORD`
   - 确保密码与你的 Neo4j 实例匹配
   - 如果看到 "NEO4J_PASSWORD environment variable is required"，则需要在 `.env` 文件中设置

3. **提取召回率低**
   - 确保使用 `agentic_v2` 方法
   - 检查 LLM API 密钥是否有效
   - 验证模型名称是否正确（gemini-flash-latest）

4. **超时错误**
   - 针对大型文档增加超时设置
   - 考虑将大型报告分块处理

5. **前端无法连接到 API**
   - 确保 API 在端口 8000 上运行
   - 检查 API 配置中的 CORS 设置

### 调试模式

启用详细日志记录：```python
import logging
logging.basicConfig(level=logging.DEBUG)

# Run extraction with debug output
result = run_agentic_v2(text, config)

开发

运行测试```bash

Unit tests

uv run pytest tests/unit

Integration tests

uv run pytest tests/integration

Specific test

uv run pytest tests/test_agentic_v2.py::test_extraction

With coverage

uv run pytest --cov=bandjacks

Frontend tests

cd ui && npm test cd ui && npm run test:coverage

### 贡献

1. 复刻（Fork）本仓库
2. 创建功能分支
3. 进行你的更改
4. 运行测试：`uv run pytest`
5. 运行代码检查：`uv run ruff check`
6. 提交拉取请求（Pull Request）

### 代码质量```bash
# Format code
uv run ruff format

# Check linting
uv run ruff check

# Type checking
uv run mypy bandjacks

许可证

[在此处填写您的许可证]

支持

快速入门: docs/QUICKSTART.md
完整安装: docs/SETUP.md
API 文档: http://localhost:8000/docs（运行时可用）
GitHub Issues: [报告错误或请求功能]

致谢

MITRE ATT&CK® 框架
D3FEND 本体
STIX 2.1 规范

`MAX_MAPPER_BATCH_SIZE` (环境变量)	10	每次 LLM 映射调用处理的跨度数（自 2026-05 起从 25 降低；云响应上限约为 800 个 token，约 12% 的大批次返回了截断的 JSON）	无
`max_spans_per_technique` (配置)	2	预过滤：每个候选技术的最佳 N 个跨度	约减少 19% 的技术，置信度更高
`enable_span_dedup` (配置)	false	在映射前移除重复的跨度文本	约减少 15% 的技术

bandjacks

最受欢迎