implementing-threat-intelligence-lifecycle-management

name	implementing-threat-intelligence-lifecycle-management
description	实现结构化威胁情报生命周期，涵盖规划、收集、处理、分析、传播和反馈阶段，为组织决策生产可操作情报。
domain	cybersecurity
subdomain	threat-intelligence
tags	["threat-intelligence","lifecycle","intelligence-cycle","collection","analysis","dissemination","strategic-intelligence","cti-program"]
version	1.0
author	mahipal
license	Apache-2.0

实现威胁情报生命周期管理

概述

威胁情报生命周期是将原始数据转化为可操作情报的结构化迭代过程。基于军事和政府机构使用的情报周期，它由六个阶段组成：指导（需求收集）、收集（数据获取）、处理（规范化和去重）、分析（情境化和评估）、传播（向相关方分发）和反馈（评估和优化）。本技能涵盖为成熟 CTI 计划构建每个阶段的工具、指标和集成点。

前置条件

• Python 3.9+ 及 pymisp、stix2、requests、pandas 库
• MISP 或 OpenCTI 作为威胁情报平台
• 票务系统（Jira、ServiceNow）用于需求管理
• SIEM 集成（Splunk、Elastic）用于指标运营化
• 理解情报分析技术（ACH、菱形模型）

核心概念

情报需求（IR）

优先情报需求（PIR）定义组织需要了解的内容。示例：哪些威胁行为者针对我们的行业？哪些漏洞正在被主动利用？我们的品牌或凭据是否在暗网上被交易？PIR 驱动收集计划并确保情报生产具有相关性。

收集管理框架

收集管理框架将情报需求映射到收集来源，跟踪收集缺口，确保覆盖整个威胁态势。来源包括 OSINT、商业 Feed、ISAC 共享、内部遥测和行业联系人的人力情报。

情报级别

战略情报为高层决策提供信息（威胁态势、风险趋势、地缘政治背景）。操作情报支持安全运营（活动跟踪、行为者 TTP、攻击时机）。战术情报实现即时防御（IOC、检测规则、黑名单）。

实践步骤

步骤 1：定义情报需求

import json
from datetime import datetime
from enum import Enum

class Priority(Enum):
    CRITICAL = 1
    HIGH = 2
    MEDIUM = 3
    LOW = 4

class IntelligenceRequirement:
    def __init__(self, requirement_id, question, priority, stakeholder,
                 intelligence_level, collection_sources=None):
        self.id = requirement_id
        self.question = question
        self.priority = priority
        self.stakeholder = stakeholder
        self.level = intelligence_level
        self.sources = collection_sources or []
        self.created = datetime.now().isoformat()
        self.status = "active"
        self.last_answered = None

    def to_dict(self):
        return {
            "id": self.id,
            "question": self.question,
            "priority": self.priority.name,
            "stakeholder": self.stakeholder,
            "intelligence_level": .level,
            : .sources,
            : .created,
            : .status,
            : .last_answered,
        }

 :
     ():
        .requirements = []

     ():
        .requirements.append(requirement)
        ()

     ():
        filtered = [r  r  .requirements  r.status == ]
         priority:
            filtered = [r  r  filtered  r.priority == priority]
         level:
            filtered = [r  r  filtered  r.level == level]
         filtered

     ():
        data = [r.to_dict()  r  .requirements]
         (output_file, )  f:
            json.dump(data, f, indent=)
        ()


mgr = RequirementsManager()
mgr.add_requirement(IntelligenceRequirement(
    , ,
    Priority.CRITICAL, , ,
    [, , ],
))
mgr.add_requirement(IntelligenceRequirement(
    , ,
    Priority.CRITICAL, , ,
    [, , , ],
))
mgr.add_requirement(IntelligenceRequirement(
    , ,
    Priority.HIGH, , ,
    [, , ],
))
mgr.add_requirement(IntelligenceRequirement(
    , ,
    Priority.HIGH, , ,
    [, , ],
))
mgr.export_requirements()

步骤 2：构建收集管道

import requests
from datetime import datetime, timedelta

class CollectionPipeline:
    def __init__(self, config):
        self.config = config
        self.collected_data = []

    def collect_cisa_kev(self):
        """收集 CISA 已知被利用漏洞目录。"""
        url = "https://www.cisa.gov/sites/default/files/feeds/known_exploited_vulnerabilities.json"
        resp = requests.get(url, timeout=30)
        if resp.status_code == 200:
            data = resp.json()
            vulns = data.get("vulnerabilities", [])
            self.collected_data.append({
                "source": "CISA KEV",
                "type": "vulnerability",
                "count": len(vulns),
                "collected_at": datetime.now().isoformat(),
                "data": vulns,
            })
            print(f"[+] CISA KEV：{len(vulns)} 个已知被利用漏洞")
            return vulns
        return []

    def collect_otx_pulses(self, api_key, days=7):
        """收集最近的 OTX 脉冲。"""
        headers = {"X-OTX-API-KEY": api_key}
        since = (datetime.now() - timedelta(days=days)).isoformat()
        url = f"https://otx.alienvault.com/api/v1/pulses/subscribed?modified_since="
        resp = requests.get(url, headers=headers, timeout=)
         resp.status_code == :
            pulses = resp.json().get(, [])
            .collected_data.append({
                : ,
                : ,
                : (pulses),
                : datetime.now().isoformat(),
            })
            ()
             pulses
         []

     ():
        
        url = 
        resp = requests.post(url, data={: , : }, timeout=)
         resp.status_code == :
            data = resp.json().get(, [])
            .collected_data.append({
                : ,
                : ,
                : (data),
                : datetime.now().isoformat(),
            })
            ()
             data
         []

     ():
        summary = {
            : (.collected_data),
            : (d.get(, )  d  .collected_data),
            : [
                {: d[], : d[], : d[]}
                 d  .collected_data
            ],
        }
         summary

pipeline = CollectionPipeline({})
pipeline.collect_cisa_kev()
pipeline.collect_abuse_ch()
(json.dumps(pipeline.get_collection_summary(), indent=))

步骤 3：处理和规范化数据

class IntelligenceProcessor:
    def __init__(self):
        self.processed_items = []
        self.dedup_hashes = set()

    def process_collection(self, raw_data, source_name):
        """规范化和去重收集到的情报。"""
        processed = []
        duplicates = 0

        for item in raw_data:
            normalized = self._normalize(item, source_name)
            if normalized:
                item_hash = self._compute_hash(normalized)
                if item_hash not in self.dedup_hashes:
                    self.dedup_hashes.add(item_hash)
                    normalized["processed_at"] = datetime.now().isoformat()
                    processed.append(normalized)
                else:
                    duplicates += 1

        self.processed_items.extend(processed)
        print(f"[+] 已从 {source_name} 处理 {len(processed)} 个条目"
              f"（已删除 {duplicates} 个重复项）")
        return processed

    def _normalize(self, item, source):
        """将条目规范化为标准格式。"""
        return {
            "source": source,
            "type": item.get("type", ),
            : item.get(, item.get(, )),
            : item.get(, ),
            : item.get(, ),
            : item.get(, []),
            : item.get(, item.get(, )),
            : item,
        }

     ():
         hashlib
        key = 
         hashlib.sha256(key.encode()).hexdigest()

processor = IntelligenceProcessor()

步骤 4：分析和生产情报

class IntelligenceAnalyzer:
    def __init__(self, requirements, processed_data):
        self.requirements = requirements
        self.data = processed_data

    def answer_requirement(self, requirement_id):
        """生产回答特定需求的情报。"""
        req = next((r for r in self.requirements if r.id == requirement_id), None)
        if not req:
            return None

        # 根据需求类型过滤相关数据
        relevant = self.data  # 实践中按需求主题过滤
        analysis = {
            "requirement_id": requirement_id,
            "question": req.question,
            "intelligence_level": req.level,
            "data_points_analyzed": len(relevant),
            "produced_at": datetime.now().isoformat(),
            "key_findings": [],
            "confidence": "medium",
            "recommendations": [],
        }
        return analysis

    def produce_daily_brief(self):
        """生产每日威胁情报简报。"""
        brief = {
            "date": datetime.now().strftime("%Y-%m-%d"),
            "total_items_processed": len(.data),
            : [],
            : [
                {: r., : r.question[:], : r.status}
                 r  .requirements  r.status == 
            ],
        }
         brief

步骤 5：传播和跟踪反馈

class IntelligenceDisseminator:
    def __init__(self):
        self.distribution_log = []

    def distribute_report(self, report, channels, classification="TLP:GREEN"):
        """通过适当渠道向相关方分发情报报告。"""
        for channel in channels:
            entry = {
                "report_id": report.get("requirement_id", "daily"),
                "channel": channel,
                "classification": classification,
                "distributed_at": datetime.now().isoformat(),
                "status": "sent",
            }
            self.distribution_log.append(entry)
            print(f"  [+] 已分发至 {channel}")

    def collect_feedback(self, report_id, stakeholder, rating, comments=""):
        """收集相关方对情报产品的反馈。"""
        feedback = {
            "report_id": report_id,
            "stakeholder": stakeholder,
            "rating": rating,  # 1-5
            "comments": comments,
            "received_at": datetime.now().isoformat(),
        }
        print(f"[+] 已收到来自 {stakeholder} 的反馈：{rating}/5")
        return feedback

    def calculate_metrics(self):
        
        metrics = {
            : (.distribution_log),
            : {},
        }
         entry  .distribution_log:
            channel = entry[]
             channel   metrics[]:
                metrics[][channel] = 
            metrics[][channel] += 
         metrics

disseminator = IntelligenceDisseminator()

验证标准

• 情报需求已定义优先级和相关方
• 收集管道从多个来源收集数据
• 处理正确去重和规范化数据
• 分析生产回答特定需求的情报
• 传播通过正确渠道触达适当相关方
• 反馈机制捕获并整合相关方输入

参考资料

• SANS：网络威胁情报生命周期
• CISA：网络安全自动化最佳实践
• MISP 项目
• STIX/TAXII 文档
• CISA 已知被利用漏洞