implementing-threat-intelligence-lifecycle-management

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

概述

威胁情报生命周期是一个结构化、迭代式的过程，用于将原始数据转化为可行动的情报。它基于军事和政府机构使用的 intelligence cycle，包含六个阶段：方向（需求收集）、收集（数据获取）、处理（规范化与去重）、分析（情境化与评估）、分发（分发给利益相关方）以及反馈（评估与改进）。本技能涵盖如何通过工具、指标和集成点构建每个阶段，以形成成熟的 CTI 项目。

何时使用

• 在您的环境中部署或配置实施威胁情报生命周期管理能力时
• 在建立与合规要求一致的安全控制时
• 在为该领域构建或改进安全架构时
• 在开展需要此实施的安全评估时

前置条件

• Python 3.9+，并安装 pymisp、stix2、requests、pandas 库
• 使用 MISP 或 OpenCTI 作为威胁情报平台
• 使用工单系统（Jira、ServiceNow）进行需求管理
• 集成 SIEM（Splunk、Elastic）以实现指标可操作化
• 理解情报分析技术（ACH、Diamond Model）

关键概念

情报需求（IR）

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

收集管理框架

收集管理框架将情报需求映射到收集来源，跟踪收集缺口，并确保覆盖整个威胁态势。来源包括 OSINT、商业情报源、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": self.level,
            "collection_sources": self.sources,
            "created": self.created,
            "status": self.status,
            "last_answered": self.last_answered,
        }

class RequirementsManager:
    def __init__(self):
        self.requirements = []

    def add_requirement(self, requirement):
        self.requirements.append(requirement)
        print(f"[+] Added IR-{requirement.id}: {requirement.question[:60]}...")

    def get_active_requirements(self, priority=None, level=None):
        filtered = [r for r in self.requirements if r.status == "active"]
        if priority:
            filtered = [r for r in filtered if r.priority == priority]
        if level:
            filtered = [r for r in filtered if r.level == level]
        return filtered

    def export_requirements(self, output_file="intelligence_requirements.json"):
        data = [r.to_dict() for r in self.requirements]
        with open(output_file, "w") as f:
            json.dump(data, f, indent=2)
        print(f"[+] Exported {len(data)} requirements to {output_file}")

# Define organizational PIRs
mgr = RequirementsManager()
mgr.add_requirement(IntelligenceRequirement(
    "PIR-001", "Which threat actors are actively targeting our sector?",
    Priority.CRITICAL, "CISO", "strategic",
    ["MITRE ATT&CK", "ISAC feeds", "Vendor reports"],
))
mgr.add_requirement(IntelligenceRequirement(
    "PIR-002", "What vulnerabilities are being actively exploited in the wild?",
    Priority.CRITICAL, "Vulnerability Management", "operational",
    ["CISA KEV", "Exploit-DB", "VulnCheck", "Shodan"],
))
mgr.add_requirement(IntelligenceRequirement(
    "PIR-003", "Are any organization credentials or data exposed on dark web?",
    Priority.HIGH, "SOC Manager", "tactical",
    ["Dark web monitoring", "Paste site monitoring", "Breach databases"],
))
mgr.add_requirement(IntelligenceRequirement(
    "PIR-004", "What are the emerging attack techniques against cloud infrastructure?",
    Priority.HIGH, "Cloud Security", "operational",
    ["ATT&CK Cloud matrix", "Vendor advisories", "ISAC bulletins"],
))
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):
        """Collect CISA Known Exploited Vulnerabilities catalog."""
        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)} known exploited vulnerabilities")
            return vulns
        return []

    def collect_otx_pulses(self, api_key, days=7):
        """Collect recent OTX pulses."""
        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={since}"
        resp = requests.get(url, headers=headers, timeout=30)
        if resp.status_code == 200:
            pulses = resp.json().get("results", [])
            self.collected_data.append({
                "source": "AlienVault OTX",
                "type": "threat_intelligence",
                "count": len(pulses),
                "collected_at": datetime.now().isoformat(),
            })
            print(f"[+] OTX: {len(pulses)} pulses in last {days} days")
            return pulses
        return []

    def collect_abuse_ch(self):
        """Collect recent malware samples from MalwareBazaar."""
        url = "https://mb-api.abuse.ch/api/v1/"
        resp = requests.post(url, data={"query": "get_recent", "selector": "time"}, timeout=30)
        if resp.status_code == 200:
            data = resp.json().get("data", [])
            self.collected_data.append({
                "source": "MalwareBazaar",
                "type": "malware_samples",
                "count": len(data),
                "collected_at": datetime.now().isoformat(),
            })
            print(f"[+] MalwareBazaar: {len(data)} recent samples")
            return data
        return []

    def get_collection_summary(self):
        summary = {
            "total_sources": len(self.collected_data),
            "total_items": sum(d.get("count", 0) for d in self.collected_data),
            "sources": [
                {"name": d["source"], "type": d["type"], "count": d["count"]}
                for d in self.collected_data
            ],
        }
        return summary

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

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

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

    def process_collection(self, raw_data, source_name):
        """Normalize and deduplicate collected intelligence."""
        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"[+] Processed {len(processed)} items from {source_name} "
              f"({duplicates} duplicates removed)")
        return processed

    def _normalize(self, item, source):
        """Normalize item to standard format."""
        return {
            "source": source,
            "type": item.get("type", "unknown"),
            "value": item.get("value", item.get("indicator", "")),
            "confidence": item.get("confidence", 50),
            "tlp": item.get("tlp", "green"),
            "tags": item.get("tags", []),
            "first_seen": item.get("first_seen", item.get("date_added", "")),
            "raw": item,
        }

    def _compute_hash(self, item):
        import hashlib
        key = f"{item['type']}:{item['value']}:{item['source']}"
        return 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):
        """Produce intelligence answering a specific requirement."""
        req = next((r for r in self.requirements if r.id == requirement_id), None)
        if not req:
            return None

        # Filter relevant data based on requirement type
        relevant = self.data  # In practice, filter by requirement topic
        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):
        """Produce daily threat intelligence brief."""
        brief = {
            "date": datetime.now().strftime("%Y-%m-%d"),
            "total_items_processed": len(self.data),
            "highlights": [],
            "active_requirements_status": [
                {"id": r.id, "question": r.question[:80], "status": r.status}
                for r in self.requirements if r.status == "active"
            ],
        }
        return brief

步骤 5：分发并跟踪反馈

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

    def distribute_report(self, report, channels, classification="TLP:GREEN"):
        """Distribute intelligence report to appropriate channels."""
        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"  [+] Distributed to {channel}")

    def collect_feedback(self, report_id, stakeholder, rating, comments=""):
        """Collect stakeholder feedback on intelligence product."""
        feedback = {
            "report_id": report_id,
            "stakeholder": stakeholder,
            "rating": rating,  # 1-5
            "comments": comments,
            "received_at": datetime.now().isoformat(),
        }
        print(f"[+] Feedback received from {stakeholder}: {rating}/5")
        return feedback

    def calculate_metrics(self):
        """Calculate CTI program performance metrics."""
        metrics = {
            "total_products_distributed": len(self.distribution_log),
            "distribution_by_channel": {},
        }
        for entry in self.distribution_log:
            channel = entry["channel"]
            if channel not in metrics["distribution_by_channel"]:
                metrics["distribution_by_channel"][channel] = 0
            metrics["distribution_by_channel"][channel] += 1
        return metrics

disseminator = IntelligenceDisseminator()

验证标准

• 已定义情报需求，并明确优先级和利益相关方
• 收集管道可从多个来源获取数据
• 处理环节可正确去重并规范化数据
• 分析可产出回答特定需求的情报
• 分发可通过正确渠道触达适当利益相关方
• 反馈机制可捕获并纳入利益相关方输入

参考资料

• SANS: Cyber Threat Intelligence Lifecycle
• CISA: Cybersecurity Automation Best Practices
• CyCognito: Threat Intelligence Lifecycle
• MISP Project
• STIX/TAXII Documentation
• CISA Known Exploited Vulnerabilities