基于 Anthropic-Cybersecurity-Skills 构建企业级威胁情报平台:MISP、OpenCTI、TheHive 与 Cortex 的集成实战指南
【免费下载链接】Anthropic-Cybersecurity-Skills817 structured cybersecurity skills for AI agents · Mapped to 6 frameworks: MITRE ATT&CK, NIST CSF 2.0, MITRE ATLAS, D3FEND, NIST AI RMF & MITRE F3 (Fight Fraud) · agentskills.io standard · Works with Claude Code, GitHub Copilot, Codex CLI, Cursor, Gemini CLI & 20+ platforms · 29 security domains · Apache 2.0项目地址: https://gitcode.com/GitHub_Trending/an/Anthropic-Cybersecurity-Skills
本篇技术指南围绕仓库中building-threat-intelligence-platform技能(SKILL.md)展开,系统讲解如何基于 Docker Compose 将 MISP、OpenCTI、TheHive、Cortex 四款开源 CTI 工具编排成统一威胁情报平台(TIP),覆盖情报采集、存储、分析、富化、响应与共享六大层级,并给出 Feed 摄取管道、Cortex 富化工作流、STIX/TAXII 互操作与分析师仪表盘的完整可运行示例。读完本文,你将具备从零设计 TIP 架构、落地数据管道并接入 SIEM 的完整实操能力。
一、什么是威胁情报平台(TIP),为什么需要它
威胁情报平台(Threat Intelligence Platform, TIP)是将多个 CTI(Cyber Threat Intelligence)工具部署并集成到一个统一系统中,用于采集、分析、富化、共享威胁情报的集中式基础设施。在 SKILL.md 中,这一技能的定位是:当安全团队需要构建集中式 CTI 平台来支撑情报运营时,提供架构设计与落地方法。
在本仓库中,该技能被归类到threat-intelligence子域,并映射到 NIST CSF 的ID.RA-01、ID.RA-05(风险识别与评估)以及DE.CM-01、DE.AE-02(连续监控与事件分析),同时映射到 MITRE ATT&CK 的T1071(应用层协议通信)、T1588.001(获取恶意软件与工具的能力)与T1591(收集受害者组织信息)。这对应 mappings/nist-csf/csf-alignment.md 中对 threat-intelligence 子域的定位:情报驱动检测,支撑安全团队理解威胁态势。
适用场景
- 正在部署或配置 TIP 能力,需要把多个情报源与工具纳入统一平台;
- 需要建立与合规要求对齐的安全控制(如 NIST CSF 的 ID/DE 类目);
- 正在建设或改进该领域的安全架构;
- 开展需要此实现支撑的安全评估。
二、前置条件与依赖
开始前需要准备以下环境与依赖(来自 SKILL.md):
| 类别 | 要求 |
|---|---|
| 容器化 | Docker 与 Docker Compose,用于部署平台各组件 |
| Python 环境 | Python 3.9+,并安装pymisp、pycti、thehive4py客户端库 |
| 数据存储 | Elasticsearch / OpenSearch 集群 |
| 消息队列 | Redis 与 RabbitMQ |
| 情报标准 | 理解 STIX 2.1 数据模型与 TAXII 2.1 传输协议 |
| 外部服务 | 用于富化的 API Key(VirusTotal、Shodan、AbuseIPDB 等) |
Python 依赖的安装方式在 process.py 中也有明确说明:pip install pymisp pycti requests。该脚本对pymisp、pycti做了可选导入保护,缺少某个库时对应组件会以None处理,便于分步搭建时先运行部分功能。
三、TIP 架构的六大核心组件
SKILL.md 定义了标准 TIP 的分层架构,从数据流入到情报流出形成完整闭环:
- 采集层(Collection Layer):从 OSINT、商业订阅与内部来源摄取 Feed;
- 存储层(Storage Layer):使用 Elasticsearch/OpenSearch 以 STIX 2.1 模式索引 CTI 数据;
- 分析层(Analysis Layer):OpenCTI 负责知识图谱分析,MISP 负责 IOC 关联;
- 富化层(Enrichment Layer):通过 Cortex analyzers 实现 IOC 自动富化;
- 响应层(Response Layer):TheHive 负责案件管理与应急响应集成;
- 共享层(Sharing Layer):TAXII 服务器负责对外情报共享。
平台集成点(Integration Points)
各组件之间的数据流动关系(来自 SKILL.md):
- MISP ↔ OpenCTI:通过 OpenCTI MISP 连接器实现双向同步;
- OpenCTI → TheHive:高置信度指标自动生成告警/案件;
- TheHive ↔ Cortex:对案件可观测对象(observables)自动分析与富化;
- 全部组件 → SIEM:通过 API 或 Kafka 将 IOC 实时推送到 Splunk / Elastic。
对应的协议与数据格式标准可参考 standards.md:
| 组件 | 协议 | 数据格式 |
|---|---|---|
| MISP | REST API | MISP JSON、STIX 2.1 |
| OpenCTI | GraphQL API | STIX 2.1 |
| TheHive | REST API | TheHive JSON |
| Cortex | REST API | Cortex Report JSON |
| Elasticsearch | REST API | JSON |
集成标准包括:MISP 通过 HTTPS + API Key 认证的 Push/Pull 同步协议;OpenCTI 连接器基于 RabbitMQ 消息队列异步处理;Cortex Analyzers 基于 Docker 且具有标准化输入输出;SIEM 集成则支持 Syslog、Kafka、REST API 或文件导出等方式。
四、Step 1:使用 Docker Compose 部署平台
SKILL.md 给出了完整的 Docker Compose 编排文件,覆盖存储、消息队列、对象存储与四个核心组件。以下是完整配置(含关键参数说明):
version: '3.8' services: # --- 存储层 --- elasticsearch: image: docker.elastic.co/elasticsearch/elasticsearch:8.12.0 environment: - discovery.type=single-node - xpack.security.enabled=false - "ES_JAVA_OPTS=-Xms2g -Xmx2g" ports: - "9200:9200" volumes: - es-data:/usr/share/elasticsearch/data redis: image: redis:7 ports: - "6379:6379" rabbitmq: image: rabbitmq:3-management ports: - "5672:5672" - "15672:15672" minio: image: minio/minio command: server /data --console-address ":9001" ports: - "9000:9000" - "9001:9001" # --- MISP --- misp: image: ghcr.io/misp/misp-docker/misp-core:latest ports: - "8443:443" environment: - MISP_ADMIN_EMAIL=admin@tip.local - MISP_BASEURL=https://localhost:8443 volumes: - misp-data:/var/www/MISP/app/files # --- OpenCTI --- opencti: image: opencti/platform:6.4.4 environment: - APP__PORT=8080 - APP__ADMIN__EMAIL=admin@tip.local - APP__ADMIN__PASSWORD=TIPAdminPassword - APP__ADMIN__TOKEN=tip-opencti-token-uuid - ELASTICSEARCH__URL=http://elasticsearch:9200 - MINIO__ENDPOINT=minio - RABBITMQ__HOSTNAME=rabbitmq - REDIS__HOSTNAME=redis ports: - "8080:8080" depends_on: - elasticsearch - redis - rabbitmq - minio # --- TheHive --- thehive: image: strangebee/thehive:5.3 environment: - TH_CORTEX_URL=http://cortex:9001 ports: - "9000:9000" depends_on: - elasticsearch # --- Cortex --- cortex: image: thehiveproject/cortex:3.1.8 ports: - "9001:9001" depends_on: - elasticsearch volumes: es-data: misp-data:参数解读:
- Elasticsearch 以
single-node单节点模式运行,ES_JAVA_OPTS建议按机器内存调整堆大小;xpack.security.enabled=false关闭安全模块以便快速联调,生产环境需启用; - OpenCTI 通过
APP__ADMIN__TOKEN设定管理令牌,后续 Python SDK 与连接器都会使用该令牌认证;ELASTICSEARCH__URL、MINIO__ENDPOINT、RABBITMQ__HOSTNAME、REDIS__HOSTNAME均指向容器网络内的服务名; - TheHive 通过
TH_CORTEX_URL指向 Cortex 服务,从而打通"案件-分析"链路; - 所有组件共享 Docker 网络,容器间用服务名互相访问,外部通过映射端口访问。
五、Step 2:配置 Feed 摄取管道
SKILL.md 提供了基于pymisp与pycti的 Feed 管理类,用于启用 OSINT 订阅源并检查 OpenCTI 连接器状态:
from pymisp import PyMISP from pycti import OpenCTIApiClient import json class TIPFeedManager: """Manage threat intelligence feed ingestion across platform components.""" def __init__(self, misp_url, misp_key, opencti_url, opencti_token): self.misp = PyMISP(misp_url, misp_key, ssl=False) self.opencti = OpenCTIApiClient(opencti_url, opencti_token) def configure_osint_feeds(self): """Enable default OSINT feeds in MISP.""" osint_feeds = [ {"name": "CIRCL OSINT", "id": 1}, {"name": "Botvrij.eu", "id": 2}, {"name": "abuse.ch URLhaus", "id": 5}, {"name": "abuse.ch Feodo Tracker", "id": 6}, ] for feed in osint_feeds: try: self.misp.enable_feed(feed["id"]) self.misp.fetch_feed(feed["id"]) print(f"[+] Enabled feed: {feed['name']}") except Exception as e: print(f"[-] Failed: {feed['name']}: {e}") def configure_opencti_connectors(self): """List and verify OpenCTI connector status.""" connectors = self.opencti.connector.list() for conn in connectors: print( f" Connector: {conn['name']} - " f"Active: {conn['active']} - " f"Type: {conn['connector_type']}" ) def sync_misp_to_opencti(self): """Verify MISP-OpenCTI sync is operational.""" # OpenCTI MISP connector handles this automatically # Check connector status connectors = self.opencti.connector.list() misp_connector = [ c for c in connectors if "misp" in c["name"].lower() ] if misp_connector: print(f"[+] MISP connector active: {misp_connector[0]['active']}") else: print("[-] MISP connector not found - configure in Docker Compose")要点说明:
- MISP 内置的 OSINT Feed(CIRCL OSINT、Botvrij.eu、abuse.ch URLhaus / Feodo Tracker)ID 是固定的,
enable_feed启用后调用fetch_feed立即拉取一次数据; - MISP → OpenCTI 的数据同步由 OpenCTI 侧的 MISP 连接器自动完成,不需要在代码中手动搬运,只需通过
connector.list()确认连接器已注册且active状态正常; - 更通用的批量启用逻辑见 process.py 的
configure_feeds()方法,它遍历self.misp.feeds()中所有enabled=False的订阅源并逐个启用,返回启用列表与总数,适合首次部署时一键初始化。
六、Step 3:构建基于 Cortex 的富化管道
Cortex 负责对 IOC 执行自动富化。 SKILL.md 给出了基于 REST API 的富化客户端,包含列出分析器、提交可观测对象、获取任务报告三个核心操作:
import requests class CortexEnrichment: """Integrate Cortex analyzers for automated enrichment.""" def __init__(self, cortex_url, cortex_key): self.url = cortex_url self.headers = {"Authorization": f"Bearer {cortex_key}"} def list_analyzers(self): """List available Cortex analyzers.""" resp = requests.get( f"{self.url}/api/analyzer", headers=self.headers, timeout=30, ) if resp.status_code == 200: analyzers = resp.json() for a in analyzers: print(f" {a['name']}: {a.get('description', '')[:60]}") return analyzers return [] def analyze_observable(self, observable_type, observable_value, analyzer_id): """Submit an observable for analysis.""" job = { "data": observable_value, "dataType": observable_type, "tlp": 2, "message": "TIP automated enrichment", } resp = requests.post( f"{self.url}/api/analyzer/{analyzer_id}/run", json=job, headers=self.headers, timeout=30, ) if resp.status_code == 200: return resp.json() return None def get_job_report(self, job_id): """Get the report for a completed analysis job.""" resp = requests.get( f"{self.url}/api/job/{job_id}/report", headers=self.headers, timeout=60, ) if resp.status_code == 200: return resp.json() return None关键参数:tlp字段遵循 TLP 数值编码(0=CLEAR、1=GREEN、2=AMBER、3=AMBER+STRICT、4=RED),此处tlp: 2表示以 TLP:AMBER 级别提交分析,确保敏感指标不外泄;dataType为可观测对象类型(如ip、domain、url、hash);提交后需用返回的job_id轮询get_job_report获取富化结果。
仓库 agent.py 中的TLP_DEFINITIONS进一步给出了完整编码表与共享范围说明:CLEAR 可无限共享、GREEN 限社区、AMBER 限组织内部、AMBER+STRICT 仅限需知范围、RED 仅限具名接收人,与 Cortex 的数值编码严格对应。
七、Step 4:实现分析师仪表盘指标
SKILL.md 提供了跨组件统计类,用于向分析师仪表盘输出实时指标:
class TIPMetrics: """Collect platform metrics for analyst dashboards.""" def __init__(self, misp, opencti): self.misp = misp self.opencti = opencti def get_platform_stats(self): """Collect statistics across all platform components.""" stats = {} # MISP stats misp_stats = self.misp.get_server_statistics() stats["misp"] = { "total_events": misp_stats.get("event_count", 0), "total_attributes": misp_stats.get("attribute_count", 0), "active_feeds": len([ f for f in self.misp.feeds() if f.get("Feed", {}).get("enabled") ]), } # OpenCTI stats via GraphQL stats["opencti"] = { "total_indicators": self.opencti.indicator.list( first=0, withPagination=True ).get("pagination", {}).get("globalCount", 0), "total_reports": self.opencti.report.list( first=0, withPagination=True ).get("pagination", {}).get("globalCount", 0), } return stats扩展实现见 process.py 的get_platform_stats():它在 MISP 侧补充了organizations(组织数)、在 OpenCTI 侧统计active_connectors(活跃连接器数),并统一附上时间戳。TIPManager还内置check_health(),对 MISP(实例版本)、OpenCTI(health.check())、TheHive(/api/status)、Cortex(/api/status)逐一探测并返回健康状态,可直接对接 template.md 中的平台状态报告模板。
命令行使用方式(来自 process.py):
python process.py --check-health --misp-url URL --misp-key KEY --opencti-url URL --opencti-token TOKEN python process.py --configure-feeds --misp-url URL --misp-key KEY python process.py --platform-stats --misp-url URL --misp-key KEY --opencti-url URL --opencti-token TOKEN结果以 JSON 输出并落盘到tip_report.json(默认文件名,可用--output覆盖)。
状态报告模板
template.md 提供了结构化的平台周报骨架,建议按以下表格组织:
| 指标类别 | 字段 |
|---|---|
| 平台健康 | 各组件状态、版本、URL |
| Feed 摄取状态 | Feed 名称、来源、状态、最近拉取时间、生成事件数 |
| 平台指标 | 事件/报告总数、指标总数、活跃 Feed、24h 富化任务数 |
| 连接器状态 | 连接器、类型(Import/Enrichment/Stream)、是否活跃、最近运行时间 |
| 建议 | 平台维护、Feed 配置改进、集成增强 |
八、STIX/TAXII 互操作与指标全生命周期管理
TIPS 的核心价值在于标准化的情报表达与交换。api-reference.md 给出了完整的 STIX 2.1 指标对象示例:
{ "type": "indicator", "spec_version": "2.1", "id": "indicator--<uuid5>", "created": "2025-01-15T10:00:00.000Z", "modified": "2025-01-15T10:00:00.000Z", "name": "Malicious IP", "pattern": "[ipv4-addr:value = '198.51.100.42']", "pattern_type": "stix", "valid_from": "2025-01-15T10:00:00.000Z", "confidence": 85, "object_marking_refs": ["marking-definition--f88d31f6-486f-44da-b317-01333bde0b82"] }TLP 标记定义 ID 对照表(STIX 2.1)
object_marking_refs引用的是标准 TLP 标记定义,ID 对照如下:
| TLP 级别 | STIX Marking Definition ID |
|---|---|
| TLP:CLEAR | marking-definition--613f2e26-407d-48c7-9eca-b8e91df99dc9 |
| TLP:GREEN | marking-definition--34098fce-860f-48ae-8e50-ebd3cc5e41da |
| TLP:AMBER | marking-definition--f88d31f6-486f-44da-b317-01333bde0b82 |
| TLP:AMBER+STRICT | marking-definition--826578e1-40a3-4b46-a8d8-b9931fdd750e |
| TLP:RED | marking-definition--5e57c739-391a-4eb3-b6be-7d15ca92d5ed |
基于 agent.py 的指标生命周期实现
仓库 agent.py 提供了完整的指标生命周期处理逻辑:
1. 指标类型自动分类:classify_indicator()通过正则识别 IP(ipv4-addr)、SHA-256 哈希(file-sha256)、MD5(file-md5)、域名(domain-name)、URL、邮箱(email-addr)等类型,并映射为对应的 STIX 模式模板:
STIX_INDICATOR_TYPES = { "ipv4-addr": "[ipv4-addr:value = '{}']", "domain-name": "[domain-name:value = '{}']", "url": "[url:value = '{}']", "file-sha256": "[file:hashes.'SHA-256' = '{}']", "file-md5": "[file:hashes.MD5 = '{}']", "email-addr": "[email-addr:value = '{}']", }2. 置信度评分:calculate_indicator_score()提供可复用的评分公式——多来源加成(每源 +15 分,上限 40)、指标老化惩罚(每天 -0.5 分,上限 30)、已确认观测加成(每条 +10 分,上限 30)、误报惩罚(每条 -15 分,上限 30),最终归一化到 0-100:
source_score = min(sources_count * 15, 40) age_penalty = min(age_days * 0.5, 30) sighting_score = min(confirmed_sightings * 10, 30) fp_penalty = min(false_positives * 15, 30) score = source_score - age_penalty + sighting_score - fp_penalty return max(0, min(100, round(score)))3. STIX Bundle 导出:build_stix_bundle()将多个指标封装为标准 STIX 2.1 Bundle,可直接用于 TAXII 发布或平台间交换;generate_tip_report()输出平台级摘要(指标总数、类型分布、平均置信度)。脚本可直接运行进行演示:python agent.py。
TAXII 2.1 共享端点操作
对外情报共享通过 TAXII 2.1 完成(api-reference.md):
# 服务发现 curl https://taxii.server.com/taxii2/ # 列出集合 curl https://taxii.server.com/taxii2/collections/ # 按类型拉取集合内对象 curl "https://taxii.server.com/taxii2/collections/{id}/objects?type=indicator" # 发布对象(STIX Bundle) curl -X POST "https://taxii.server.com/taxii2/collections/{id}/objects" \ -H "Content-Type: application/stix+json;version=2.1" \ -d @bundle.jsonOpenCTI GraphQL 与 MISP REST 写入示例
OpenCTI 以 GraphQL 为核心 API,创建指标:
mutation { indicatorAdd(input: { name: "Malicious IP" pattern: "[ipv4-addr:value = '198.51.100.42']" pattern_type: "stix" x_opencti_score: 80 }) { id standard_id } }MISP 以 REST API 添加属性(attribute):
curl -X POST "https://misp/attributes/add/EVENT_ID" \ -H "Authorization: MISP_KEY" \ -H "Content-Type: application/json" \ -d '{"type":"ip-dst","value":"198.51.100.42","category":"Network activity","to_ids":true}'其中to_ids: true表示该属性用于 IDS 告警匹配,会同步进 SIEM 检测规则。
九、三类核心工作流
workflows.md 定义了 TIP 运营的三种典型工作流,可作为架构设计与排障的路线图:
工作流 1:端到端情报管道
[External Feeds] --> [MISP] --> [OpenCTI] --> [Enrichment (Cortex)] --> [SIEM/TheHive] | | | | | v v v v v OSINT/Commercial Correlate Knowledge Graph VT/Shodan/AIPDB Alerts/Cases外部 OSINT/商业情报源进入 MISP 完成关联分析,同步到 OpenCTI 构建知识图谱,经 Cortex 调用 VirusTotal / Shodan / AbuseIPDB 富化,最终推送到 SIEM 生成检测规则或进入 TheHive 生成案件。
工作流 2:事件到情报的反馈闭环
[SOC Alert] --> [TheHive Case] --> [Cortex Analysis] --> [IOC Extraction] | v [MISP Event Creation] | v [OpenCTI Knowledge Update] | v [Updated Detections --> SIEM]SOC 告警进入 TheHive 开案,Cortex 分析提取 IOC,回写 MISP 创建事件,同步更新 OpenCTI 知识库,最终反哺 SIEM 检测规则——形成"检测 → 分析 → 情报沉淀 → 再检测"的闭环,这也是情报体系自我进化能力的来源。
工作流 3:平台健康监控
[Prometheus/Grafana] --> [Component Health] --> [Feed Status] --> [Alert on Failure] | | v v [ES Cluster Health] [Connector Status]通过 Prometheus/Grafana 采集 Elasticsearch 集群健康、连接器状态、Feed 状态等指标,失败时自动告警。这与 process.py 的check_health()能力互补——脚本侧重部署期的连通性验证,监控体系负责运行期的持续观测。
十、验证标准与上线检查清单
部署完成后,对照 SKILL.md 的验证标准逐项确认:
- ✅ 所有平台组件(MISP、OpenCTI、TheHive、Cortex)均已部署并可访问;
- ✅ MISP ↔ OpenCTI 双向同步正常运作;
- ✅ 至少 3 个 OSINT Feed 持续摄取数据;
- ✅ Cortex analyzers 已配置并能返回富化结果;
- ✅ 平台指标仪表盘显示实时统计;
- ✅ STIX/TAXII 导出可用于情报共享。
十一、从技能到落地:总结
构建威胁情报平台并不是简单地把四个开源工具"装在一起",而是通过分层架构、标准数据模型与自动化管道将它们组织成一条可运转的情报生产线。本文所讲的六层架构(采集、存储、分析、富化、响应、共享)、三组集成点、四步落地流程与三类运营工作流,构成了一个完整的 TIP 建设方法论:
- 架构先行:明确各组件职责与数据流,用 Docker Compose 一键拉起基础设施;
- 管道为王:Feed 摄取与 Cortex 富化管道决定情报的时效性与质量;
- 标准贯穿:STIX 2.1 + TAXII 2.1 保证跨平台互操作,TLP 机制控制情报共享边界;
- 闭环进化:事件到情报的反馈环让平台随实战不断积累知识。
进一步阅读本仓库的延伸材料:agent.py(STIX 指标生命周期与评分实现)、process.py(平台运维管理脚本)、standards.md(标准与框架对照)、workflows.md(工作流设计)以及 template.md(状态报告模板)。关于该技能在仓库框架体系中的定位,可参考 csf-alignment.md 中 threat-intelligence 子域的 NIST CSF 映射(ID.RA、DE.AE)与 README.md 的整体框架说明。
【免费下载链接】Anthropic-Cybersecurity-Skills817 structured cybersecurity skills for AI agents · Mapped to 6 frameworks: MITRE ATT&CK, NIST CSF 2.0, MITRE ATLAS, D3FEND, NIST AI RMF & MITRE F3 (Fight Fraud) · agentskills.io standard · Works with Claude Code, GitHub Copilot, Codex CLI, Cursor, Gemini CLI & 20+ platforms · 29 security domains · Apache 2.0项目地址: https://gitcode.com/GitHub_Trending/an/Anthropic-Cybersecurity-Skills
创作声明:本文部分内容由AI辅助生成(AIGC),仅供参考