implementing-taxii-server-with-opentaxii

name	implementing-taxii-server-with-opentaxii
description	部署和配置 OpenTAXII 服务器，使用 TAXII 2.1 协议共享和消费 STIX 格式的网络威胁情报，实现组织间的自动化指标交换。
domain	cybersecurity
subdomain	threat-intelligence
tags	["taxii","stix","opentaxii","threat-sharing","cti","indicator-exchange","taxii-server","automation"]
version	1.0
author	mahipal
license	Apache-2.0

使用 OpenTAXII 实现 TAXII 服务器

概述

TAXII（可信自动化情报信息交换）是 OASIS 标准协议，用于通过 HTTPS 交换网络威胁情报。OpenTAXII 是 EclecticIQ 开发的开源 TAXII 服务器实现，支持 TAXII 1.x；OASIS cti-taxii-server 提供 TAXII 2.1 参考实现。本技能涵盖部署 TAXII 服务器，配置威胁情报 Feed 集合，发布 STIX 2.1 Bundle，以及与 SIEM/SOAR 平台集成实现自动化指标摄入。

前置条件

• Python 3.9+ 及 medallion、stix2、taxii2-client、opentaxii、cabby 库
• Docker 和 Docker Compose 用于容器化部署
• 理解 STIX 2.1 对象（Indicator、Malware、Attack Pattern、Relationship）
• 熟悉 REST API 和 HTTPS 配置
• 生产环境部署所需 TLS 证书

核心概念

TAXII 2.1 架构

TAXII 2.1 定义三种服务：发现（查找可用 API 根）、API 根（集合的入口点）和集合（CTI 对象的存储库）。集合支持两种访问模式：集合端点允许消费者轮询对象，状态端点跟踪添加操作的结果。TAXII 使用带 application/taxii+json;version=2.1 的 HTTP 内容协商。

共享模型

TAXII 支持集线器和辐射（中心服务器向消费者分发）、点对点（合作伙伴之间双向共享）和源-订阅（生产者发布，消费者订阅）模型。每个集合可以有只读、只写或读写访问控制。

STIX 2.1 内容

TAXII 传输包含结构化威胁信息对象的 STIX 2.1 Bundle：指标（检测模式）、观测数据、恶意软件、攻击模式、威胁行为者、入侵集合、活动、关系和目击记录。每个对象都有唯一的 STIX ID、创建/修改时间戳和可选的 TLP 标记定义。

实践步骤

步骤 1：使用 Medallion 部署 TAXII 2.1 服务器

# 安装 medallion（OASIS 参考实现）
# pip install medallion

# medallion_config.json
import json

config = {
    "backend": {
        "module_class": "MemoryBackend",
        "filename": "taxii_data.json"
    },
    "users": {
        "admin": "admin_password_change_me",
        "analyst": "analyst_password_change_me",
        "readonly": "readonly_password_change_me"
    },
    "taxii": {
        "max_content_length": 10485760
    }
}

# 创建初始数据存储
taxii_data = {
    "discovery": {
        "title": "威胁情报 TAXII 服务器",
        "description": "用于共享 CTI 指标的 TAXII 2.1 服务器",
        "contact": "[e-mail]",
        "default": "https://taxii.organization.com/api/",
        "api_roots": ["https://taxii.organization.com/api/"]
    },
    "api_roots": {
        "api": {
            "title": "威胁情报 API 根",
            "description": "威胁情报共享的主要 API 根",
            "versions": ["application/taxii+json;version=2.1"],
            "max_content_length": 10485760,
            "collections": {
                "malware-iocs": {
                    "id": "91a7b528-80eb-42ed-a74d-c6fbd5a26116",
                    : ,
                    : ,
                    : ,
                    : ,
                    : []
                },
                : {
                    : ,
                    : ,
                    : ,
                    : ,
                    : ,
                    : []
                },
                : {
                    : ,
                    : ,
                    : ,
                    : ,
                    : ,
                    : []
                }
            }
        }
    }
}

 (, )  f:
    json.dump(config, f, indent=)
 (, )  f:
    json.dump(taxii_data, f, indent=)
()

步骤 2：Docker 部署

# docker-compose.yml
version: '3.8'
services:
  taxii-server:
    image: python:3.11-slim
    container_name: taxii-server
    working_dir: /app
    volumes:
      - ./medallion_config.json:/app/medallion_config.json
      - ./taxii_data.json:/app/taxii_data.json
      - ./certs:/app/certs
    ports:
      - "6100:6100"
    command: >
      bash -c "pip install medallion &&
      medallion --host 0.0.0.0 --port 6100
      --config /app/medallion_config.json"
    restart: unless-stopped
    healthcheck:
      test: ["CMD", "curl", "-f", "http://localhost:6100/taxii2/"]
      interval: 30s
      timeout: 10s
      retries: 3

步骤 3：向集合发布 STIX 2.1 对象

from stix2 import Indicator, Malware, Relationship, Bundle, TLP_WHITE
from taxii2client.v21 import Server, Collection, as_pages
import json
from datetime import datetime

class TAXIIPublisher:
    def __init__(self, server_url, username, password):
        self.server = Server(
            server_url,
            user=username,
            password=password,
        )

    def list_collections(self):
        """列出所有可用集合。"""
        api_root = self.server.api_roots[0]
        for collection in api_root.collections:
            print(f"  [{collection.id}] {collection.title} "
                  f"（读={collection.can_read}, 写={collection.can_write}）")
        return api_root.collections

    def publish_indicators(self, collection_id, indicators):
        """将 STIX 指标发布到 TAXII 集合。"""
        api_root = self.server.api_roots[0]
        collection = Collection(
            f"{api_root.url}collections/{collection_id}/",
            user=self.server._user,
            password=self.server._password,
        )
        bundle = Bundle(objects=indicators)
        response = collection.add_objects(bundle.serialize())
        print()
        ()
         response

     ():
        
        malware = Malware(
            name=,
            description=
                        ,
            malware_types=[, ],
            is_family=,
            object_marking_refs=[TLP_WHITE],
        )

        indicator_hash = Indicator(
            name=,
            description=,
            pattern=
                    ,
            pattern_type=,
            valid_from=datetime(, , ),
            indicator_types=[],
            object_marking_refs=[TLP_WHITE],
        )

        indicator_domain = Indicator(
            name=,
            description=,
            pattern=
                    ,
            pattern_type=,
            valid_from=datetime(, , ),
            indicator_types=[],
            object_marking_refs=[TLP_WHITE],
        )

        rel = Relationship(
            relationship_type=,
            source_ref=indicator_hash.,
            target_ref=malware.,
        )

         [malware, indicator_hash, indicator_domain, rel]

publisher = TAXIIPublisher(
    ,
    , 
)
collections = publisher.list_collections()
indicators = publisher.create_malware_indicators()
publisher.publish_indicators(, indicators)

步骤 4：从 TAXII 集合消费情报

from taxii2client.v21 import Server, Collection, as_pages
import json

class TAXIIConsumer:
    def __init__(self, server_url, username, password):
        self.server = Server(server_url, user=username, password=password)

    def poll_collection(self, collection_id, added_after=None):
        """轮询集合获取新 STIX 对象。"""
        api_root = self.server.api_roots[0]
        collection = Collection(
            f"{api_root.url}collections/{collection_id}/",
            user=self.server._user,
            password=self.server._password,
        )

        kwargs = {}
        if added_after:
            kwargs["added_after"] = added_after

        all_objects = []
        for bundle in as_pages(collection.get_objects, per_request=50, **kwargs):
            objects = json.loads(bundle).get("objects", [])
            all_objects.extend(objects)

        indicators = [o for o in all_objects if o.get("type") == "indicator"]
        malware = [o for o in all_objects if o.get("type") == "malware"]
        relationships = [o for o in all_objects if o.get("type") == "relationship"]

        print(
              
              )
         all_objects

     ():
        
        iocs = []
         obj  stix_objects:
             obj.get() == :
                pattern = obj.get(, )
                iocs.append({
                    : obj.get(),
                    : obj.get(, ),
                    : pattern,
                    : obj.get(, ),
                    : obj.get(, []),
                    : obj.get(, ),
                })
         iocs

consumer = TAXIIConsumer(
    ,
    , 
)
objects = consumer.poll_collection()
iocs = consumer.extract_iocs_for_siem(objects)

步骤 5：与 SIEM/SOAR 集成

import requests

def push_to_splunk(iocs, splunk_url, hec_token):
    """通过 HEC 将提取的 IOC 推送到 Splunk。"""
    headers = {"Authorization": f"Splunk {hec_token}"}
    for ioc in iocs:
        event = {
            "event": ioc,
            "sourcetype": "stix:indicator",
            "source": "taxii-server",
            "index": "threat_intel",
        }
        resp = requests.post(
            f"{splunk_url}/services/collector/event",
            headers=headers,
            json=event,
            verify=False,
        )
        if resp.status_code != 200:
            print(f"[-] Splunk HEC 错误：{resp.text}")
    print(f"[+] 已向 Splunk 推送 {len(iocs)} 个 IOC")

def push_to_elasticsearch(iocs, es_url, index="threat-intel"):
    """将 IOC 推送到 Elasticsearch。"""
    for ioc in iocs:
        resp = requests.post(
            f"{es_url}/{index}/_doc",
            json=ioc,
            headers={"Content-Type": "application/json"},
        )
        if resp.status_code not in (200, 201):
            ()
    ()

验证标准

• TAXII 2.1 服务器部署并可通过 HTTPS 访问
• 创建具有适当读/写权限的集合
• STIX 2.1 Bundle 成功发布到集合
• 消费者可以使用过滤功能轮询和检索对象
• IOC 提取并转发到 SIEM 平台
• 正确执行身份验证和授权

参考资料

• TAXII 2.1 规范
• OASIS CTI 文档
• EclecticIQ OpenTAXII
• cti-taxii-server（Medallion）
• taxii2-client Python 库