| name | building-ioc-enrichment-pipeline-with-opencti |
| description | OpenCTI 是一个以 STIX 2.1 为原生数据模型的开源网络威胁情报知识管理平台。本技能涵盖使用 OpenCTI 连接器生态系统构建自动化 IOC 富化流水线，通过 VirusTotal、Shodan、AbuseIPDB、GreyNoise 等来源对指标进行富化。 |
| domain | cybersecurity |
| subdomain | threat-intelligence |
| tags | ["threat-intelligence","cti","ioc","mitre-attack","stix","opencti","enrichment","virustotal"] |
| version | 1.0 |
| author | mahipal |
| license | Apache-2.0 |
使用 OpenCTI 构建 IOC 富化流水线
概述
OpenCTI 是一个以 STIX 2.1 为原生数据模型的开源网络威胁情报知识管理平台。本技能涵盖使用 OpenCTI 连接器生态系统构建自动化 IOC 富化流水线，通过 VirusTotal、Shodan、AbuseIPDB、GreyNoise 等来源对指标进行富化。该流水线自动对新摄入的指标进行富化，将其与已知威胁行为者和攻击活动关联，并为分析师优先排序进行评分。
前置条件
- 用于部署 OpenCTI 的 Docker 和 Docker Compose
- Python 3.9+，安装
pycti 库
- 富化服务 API 密钥：VirusTotal、Shodan、AbuseIPDB、GreyNoise
- 了解 STIX 2.1 数据模型和关系
- OpenCTI 后端需要 ElasticSearch 或 OpenSearch
- 连接器消息队列需要 RabbitMQ 或 Redis
核心概念
OpenCTI 架构
OpenCTI 使用 GraphQL API 前端，以 ElasticSearch 作为存储后端，以 Redis/RabbitMQ 用于连接器通信。数据以 STIX 2.1 对象和关系的形式原生存储。连接器分为以下类别：外部导入（推送摄取）、内部导入（文件解析）、内部富化（上下文添加）和流式处理（实时导出）。
富化连接器模型
内部富化连接器在创建新可观测对象时自动触发，或由分析师手动触发。每个连接器接收 STIX 对象、查询外部服务，并返回 STIX 2.1 bundle，以附加的上下文、标签和关系扩充原始可观测对象。
置信度评分
OpenCTI 对指标使用 0-100 置信度等级。富化连接器可根据外部验证更新置信度分数：VirusTotal 检测率、Shodan 暴露数据、AbuseIPDB 报告数量和 GreyNoise 分类结果。
实践步骤
步骤 1：使用 Docker Compose 部署 OpenCTI
version: '3'
services:
opencti:
image: opencti/platform:6.4.4
environment:
- APP__PORT=8080
- APP__ADMIN__EMAIL=admin@opencti.io
- APP__ADMIN__PASSWORD=ChangeMeNow
- APP__ADMIN__TOKEN=your-admin-token-uuid
- ELASTICSEARCH__URL=http://elasticsearch:9200
- MINIO__ENDPOINT=minio
- RABBITMQ__HOSTNAME=rabbitmq
ports:
- "8080:8080"
depends_on:
- elasticsearch
- minio
- rabbitmq
- redis
connector-virustotal:
image: opencti/connector-virustotal:6.4.4
environment:
- OPENCTI_URL=http://opencti:8080
- OPENCTI_TOKEN=your-admin-token-uuid
- CONNECTOR_ID=connector-virustotal-id
- CONNECTOR_NAME=VirusTotal
- CONNECTOR_SCOPE=StixFile,Artifact,IPv4-Addr,Domain-Name,Url
- CONNECTOR_AUTO=true
步骤 2：构建自定义富化连接器
import os
from pycti import OpenCTIConnectorHelper, get_config_variable
from stix2 import (
Bundle, Indicator, Note, Relationship,
IPv4Address, DomainName
)
import requests
class CustomEnrichmentConnector:
def __init__(self):
config = {
"opencti": {
"url": os.environ.get("OPENCTI_URL"),
"token": os.environ.get("OPENCTI_TOKEN"),
},
"connector": {
"id": os.environ.get("CONNECTOR_ID"),
"name": "CustomEnrichment",
"scope": "IPv4-Addr,Domain-Name,Url",
"auto": True,
"type": "INTERNAL_ENRICHMENT",
},
}
self.helper = OpenCTIConnectorHelper(config)
self.helper.listen(self._process_message)
def _process_message(self, data):
entity_id = data["entity_id"]
stix_object = self.helper.api.stix_cyber_observable.read(id=entity_id)
if not stix_object:
return "未找到可观测对象"
observable_type = stix_object["entity_type"]
observable_value = stix_object.get("value", )
enrichment_results = []
observable_type == :
enrichment_results = ._enrich_ip(observable_value, entity_id)
observable_type == :
enrichment_results = ._enrich_domain(observable_value, entity_id)
enrichment_results:
bundle = Bundle(objects=enrichment_results, allow_custom=)
.helper.send_stix2_bundle(bundle.serialize())
():
objects = []
:
gn_response = requests.get(
,
headers={: os.environ.get()},
timeout=,
)
gn_response.status_code == :
gn_data = gn_response.json()
classification = gn_data.get(, )
noise = gn_data.get(, )
riot = gn_data.get(, )
note_content = (
)
note = Note(
content=note_content,
object_refs=[entity_id],
abstract=,
allow_custom=,
)
objects.append(note)
classification == :
.helper.api.stix_cyber_observable.add_label(
=entity_id, label_name=
)
riot:
.helper.api.stix_cyber_observable.add_label(
=entity_id, label_name=
)
Exception e:
.helper.log_error()
objects
():
objects = []
:
st_response = requests.get(
,
headers={: os.environ.get()},
timeout=,
)
st_response.status_code == :
st_data = st_response.json()
current_dns = st_data.get(, {})
a_records = [
r.get() r current_dns.get(, {}).get(, [])
]
note_content = (
)
note = Note(
content=note_content,
object_refs=[entity_id],
abstract=,
allow_custom=,
)
objects.append(note)
Exception e:
.helper.log_error()
objects
__name__ == :
connector = CustomEnrichmentConnector()
验收标准
- OpenCTI 实例成功部署并可访问
- VirusTotal 和 Shodan 连接器自动富化新指标
- 自定义连接器处理 GreyNoise 和 SecurityTrails 富化
- 置信度分数随富化结果更新
- 标签根据分类结果自动应用
- STIX bundle 正确通过连接器通信传递
参考资料
