processing-stix-taxii-feeds

# 处理 STIX/TAXII 情报源

## 使用场景

在以下情况使用本技能：
- 从政府情报源（CISA AIS、FS-ISAC）或商业提供商接入新的 TAXII 2.1 集合
- 在导入前验证摄入的 STIX 包是否符合 OASIS STIX 2.1 规范
- 构建解析 STIX 关系对象以重建活动上下文的自动化管道

**不适用**于 Recorded Future JSON、CrowdStrike IOC 列表等专有厂商情报格式（需要厂商专属解析器而非 STIX 处理）。

## 前置条件

- Python 3.9+，安装 `stix2` 库（pip install stix2）和 `taxii2-client` 库
- 可访问 TAXII 2.1 服务器端点，具备有效凭证
- 目标 TIP 或 SIEM 具备导入 API（MISP、OpenCTI 或 Splunk ES）

## 工作流

### 步骤 1：发现 TAXII 服务器集合

```python
from taxii2client.v21 import Server, as_pages

server = Server("https://cti.example.com/taxii/",
                user="apiuser", password="apikey")
api_root = server.api_roots[0]
for collection in api_root.collections:
    print(collection.id, collection.title, collection.can_read)
```

选择与您威胁画像相关的集合。CISA AIS 提供按行业分段的集合（金融、能源、医疗）。

### 步骤 2：分页获取 STIX 包

```python
from taxii2client.v21 import Collection
from datetime import datetime, timedelta, timezone

collection = Collection(
    "https://cti.example.com/taxii/api1/collections/<id>/objects/",
    user="apiuser", password="apikey")

# 仅获取过去 24 小时内新增的对象
added_after = datetime.now(timezone.utc) - timedelta(hours=24)
for bundle_page in as_pages(collection.get_objects,
                             added_after=added_after, per_request=100):
    process_bundle(bundle_page)
```

### 步骤 3：解析和验证 STIX 对象

```python
import stix2

def process_bundle(bundle_dict):
    bundle = stix2.parse(bundle_dict, allow_custom=True)
    for obj in bundle.objects:
        if obj.type == "indicator":
            validate_indicator(obj)
        elif obj.type == "threat-actor":
            upsert_threat_actor(obj)
        elif obj.type == "relationship":
            link_objects(obj)

def validate_indicator(indicator):
    required = ["id", "type", "spec_version", "created",
                "modified", "pattern", "pattern_type", "valid_from"]
    for field in required:
        if not hasattr(indicator, field):
            raise ValueError(f"缺少必填字段: {field}")
    # 检查置信度范围
    if hasattr(indicator, "confidence"):
        assert 0 <= indicator.confidence <= 100
```

### 步骤 4：将对象路由到消费平台

将 STIX 对象类型映射到目标系统：
- `indicator` 对象 → SIEM 查询表和防火墙封锁列表
- `malware` 对象 → EDR 威胁情报库
- `threat-actor` / `campaign` 对象 → TIP 用于分析师上下文
- `course-of-action` 对象 → 安全团队 Wiki 或 SOAR 剧本触发器

使用 TLP 标记定义（Marking Definition）强制执行共享限制：
```python
for marking in obj.get("object_marking_refs", []):
    if "tlp-red" in marking:
        route_to_restricted_platform_only(obj)
```

### 步骤 5：回发至 TAXII（双向共享）

```python
# 将经验证的本地情报添加回共享集合
new_indicator = stix2.Indicator(
    name="恶意 C2 域名",
    pattern="[domain-name:value = 'evil-c2.example.com']",
    pattern_type="stix",
    valid_from="2025-01-15T00:00:00Z",
    confidence=80,
    labels=["malicious-activity"],
    object_marking_refs=["marking-definition--34098fce-860f-479c-ae..."]  # TLP:GREEN
)
collection.add_objects(stix2.Bundle(new_indicator))
```

## 核心概念

| 术语 | 定义 |
|------|-----------|
| **STIX Bundle** | 顶层 STIX 容器对象（类型: "bundle"），包含任意数量的 STIX 域对象（SDO）和 STIX 关系对象（SRO） |
| **SDO** | STIX 域对象（STIX Domain Object）——核心情报类型：indicator、threat-actor、malware、campaign、attack-pattern、course-of-action |
| **SRO** | STIX 关系对象（STIX Relationship Object）——用带标签的关系（如 "uses"、"attributed-to"、"indicates"）连接两个 SDO |
| **模式语言** | 指标条件的 STIX 模式语法：`[network-traffic:dst_port = 443 AND ipv4-addr:value = '10.0.0.1']` |
| **标记定义** | 编码 TLP 或声明共享限制的 STIX 对象 |
| **added_after** | TAXII 2.1 过滤参数（RFC 3339 时间戳），用于增量轮询新对象 |

## 工具与系统

- **stix2（Python）**：用于创建、解析和验证 STIX 2.0/2.1 对象的官方 OASIS Python 库
- **taxii2-client（Python）**：用于 TAXII 2.0/2.1 服务器发现、集合枚举和对象检索的客户端库
- **MISP**：具有原生 TAXII 2.1 服务端和客户端的开源 TIP；MISP-TAXII-Server 插件用于发布 MISP 事件
- **OpenCTI**：内置 TAXII 2.1 连接器的 CTI 平台；原生支持 STIX 2.1 导入/导出
- **Cabby**：用于仍使用 TAXII 1.1 的旧版政府情报源的遗留 Python TAXII 1.x 客户端

## 常见陷阱

- **忽略 `spec_version` 字段**：STIX 2.0 和 2.1 具有不兼容的模式（2.1 新增了 `confidence`、包级别的 `object_marking_refs`）。解析前务必检查 `spec_version`。
- **未处理分页**：TAXII 服务器每次请求上限为 100-1000 个对象。遗漏分页（通过 `next` 链接头）会导致数据静默丢失。
- **`added_after` 时钟偏差**：服务端和客户端时间不一致会导致在间隔边界处遗漏对象。统一使用 UTC 并添加 5 分钟重叠窗口。
- **原始 STIX 块未建立索引即存储**：将包作为不透明 JSON 存储会阻止按指标类型或活动查询。应解析为关系型或图形数据库。
- **无意共享 TLP:RED 内容**：自动化管道在路由到任何具有广泛分析师访问权限的共享平台或 SIEM 前，必须过滤标记定义。