使用 OpenCTI 建構 IOC 情資擴增管道

使用 OpenCTI 建構 IOC 情資擴增管道 是做什麼的？

概述

OpenCTI 是一個開放原始碼的網路威脅情資知識管理平台，以 STIX 2.1 作為其原生資料模型。本技能涵蓋如何使用 OpenCTI 的連接器生態系統，建構自動化的 IOC 情資擴增管道，從 VirusTotal、Shodan、AbuseIPDB、GreyNoise 等來源為指標加入上下文資訊。此管道會自動擴增新匯入的指標，將其與已知威脅行為者和戰術活動進行關聯，並依據信心分數進行分析師優先級排序。

使用時機

• 在您的環境中部署或設定使用 OpenCTI 建構 IOC 情資擴增管道的功能時
• 建立符合合規要求的安全控制措施時
• 為此領域建構或改善安全架構時
• 在需要此實作的安全評估過程中

前置需求

• 用於部署 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 物件及其關係的形式儲存。連接器分為以下類別：外部匯入（來源摘要匯入）、內部匯入（檔案解析）、內部擴增（上下文加入）以及串流（即時匯出）。

擴增連接器模型

內部擴增連接器會在建立新的 Observable（可觀察物件）時自動觸發，或由分析師手動觸發。每個連接器接收 STIX 物件，查詢外部服務，並回傳 STIX 2.1 套件，其中包含為原始 Observable 附加的額外上下文、標籤以及關係。

信心分數評分

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
      - [email protected]
      - 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
      - VIRUSTOTAL_TOKEN=your-vt-api-key
      - VIRUSTOTAL_MAX_TLP=TLP:AMBER

  connector-shodan:
    image: opencti/connector-shodan:6.4.4
    environment:
      - OPENCTI_URL=http://opencti:8080
      - OPENCTI_TOKEN=your-admin-token-uuid
      - CONNECTOR_ID=connector-shodan-id
      - CONNECTOR_NAME=Shodan
      - CONNECTOR_SCOPE=IPv4-Addr
      - CONNECTOR_AUTO=true
      - SHODAN_TOKEN=your-shodan-api-key
      - SHODAN_MAX_TLP=TLP:AMBER

  connector-abuseipdb:
    image: opencti/connector-abuseipdb:6.4.4
    environment:
      - OPENCTI_URL=http://opencti:8080
      - OPENCTI_TOKEN=your-admin-token-uuid
      - CONNECTOR_ID=connector-abuseipdb-id
      - CONNECTOR_NAME=AbuseIPDB
      - CONNECTOR_SCOPE=IPv4-Addr
      - CONNECTOR_AUTO=true
      - ABUSEIPDB_API_KEY=your-abuseipdb-key

步驟 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 not found"

        observable_type = stix_object["entity_type"]
        observable_value = stix_object.get("value", "")

        enrichment_results = []

        if observable_type == "IPv4-Addr":
            enrichment_results = self._enrich_ip(observable_value, entity_id)
        elif observable_type == "Domain-Name":
            enrichment_results = self._enrich_domain(observable_value, entity_id)

        if enrichment_results:
            bundle = Bundle(objects=enrichment_results, allow_custom=True)
            self.helper.send_stix2_bundle(bundle.serialize())

        return "Enrichment completed"

    def _enrich_ip(self, ip_address, entity_id):
        """使用 GreyNoise、AbuseIPDB 上下文擴增 IP 地址。"""
        objects = []

        # GreyNoise Community API
        try:
            gn_response = requests.get(
                f"https://api.greynoise.io/v3/community/{ip_address}",
                headers={"key": os.environ.get("GREYNOISE_API_KEY")},
                timeout=30,
            )
            if gn_response.status_code == 200:
                gn_data = gn_response.json()
                classification = gn_data.get("classification", "unknown")
                noise = gn_data.get("noise", False)
                riot = gn_data.get("riot", False)

                note_content = (
                    f"## GreyNoise 擴增結果\n"
                    f"- 分類：{classification}\n"
                    f"- 網際網路噪音：{noise}\n"
                    f"- RIOT（良性服務）：{riot}\n"
                    f"- 名稱：{gn_data.get('name', 'N/A')}\n"
                    f"- 最後出現時間：{gn_data.get('last_seen', 'N/A')}"
                )

                note = Note(
                    content=note_content,
                    object_refs=[entity_id],
                    abstract=f"GreyNoise: {classification}",
                    allow_custom=True,
                )
                objects.append(note)

                # 根據分類加入標籤
                if classification == "malicious":
                    self.helper.api.stix_cyber_observable.add_label(
                        id=entity_id, label_name="greynoise:malicious"
                    )
                elif riot:
                    self.helper.api.stix_cyber_observable.add_label(
                        id=entity_id, label_name="greynoise:benign-service"
                    )

        except Exception as e:
            self.helper.log_error(f"GreyNoise 擴增失敗：{e}")

        return objects

    def _enrich_domain(self, domain, entity_id):
        """使用 WHOIS 和 DNS 上下文擴增域名。"""
        objects = []

        try:
            # 使用 SecurityTrails API 進行域名擴增
            st_response = requests.get(
                f"https://api.securitytrails.com/v1/domain/{domain}",
                headers={"APIKEY": os.environ.get("SECURITYTRAILS_API_KEY")},
                timeout=30,
            )
            if st_response.status_code == 200:
                st_data = st_response.json()
                current_dns = st_data.get("current_dns", {})

                a_records = [
                    r.get("ip") for r in current_dns.get("a", {}).get("values", [])
                ]

                note_content = (
                    f"## SecurityTrails 擴增結果\n"
                    f"- A 記錄：{', '.join(a_records)}\n"
                    f"- Alexa 排名：{st_data.get('alexa_rank', 'N/A')}\n"
                    f"- 主機名稱：{st_data.get('hostname', 'N/A')}"
                )

                note = Note(
                    content=note_content,
                    object_refs=[entity_id],
                    abstract=f"SecurityTrails: {domain}",
                    allow_custom=True,
                )
                objects.append(note)

        except Exception as e:
            self.helper.log_error(f"SecurityTrails 擴增失敗：{e}")

        return objects


if __name__ == "__main__":
    connector = CustomEnrichmentConnector()