building-ioc-enrichment-pipeline-with-opencti

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

# docker-compose.yml（核心服务）
version: '3'
services:
  opencti:
    image: opencti/platform:6.4.4
    environment:
      - APP__PORT=8080
      - APP__ADMIN__EMAIL=[e-mail]
      - 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 正确通过连接器通信传递

参考资料

• OpenCTI 文档
• OpenCTI GitHub
• pycti Python 库
• OpenCTI 连接器库
• GreyNoise API