White-Hat-007/APT-Intelligence-Engine

# 高级持续性威胁 (APT) 情报引擎 ## 🛑 执行摘要 **高级持续性威胁 (APT) 情报引擎**是一个高性能、模块化的框架，专为多模态安全遥测的实时摄取、标准化和行为分析而设计。该引擎超越了静态的入侵指标，利用对抗性行为指纹、MITRE ATT&CK 映射以及无监督机器学习（K-Means 聚类），来识别、跟踪和关联跨不同数据源的复杂入侵活动。 该引擎采用安全第一、可扩展的架构开发，可与企业的 SOC 基础设施（Splunk、ELK、Kafka）无缝集成，提供即时的战术洞察和战略情报报告。 ## 🧬 核心架构与组件 该引擎基于流水线架构运行，包含四个主要阶段：摄取、映射、分析和报告。 ``` graph TD A[Telemetry Ingestor
Kafka / Splunk / Syslog / WebSocket] --> B[Adversarial Mapping Layer
mitre_mapper.py / technique_inference] B --> C[Analytics & Fingerprinting
fingerprint_engine.py / similarity_engine.py] C --> D[Clustering Engine
clustering_engine.py / K-Means] C --> E[Graph Builder
graph_builder.py / Attack Path Network] D --> F[Intelligence Reporting
intelligence_report.py] E --> F F --> G[Strategic Intelligence Reports] style A fill:#2b5c8f,stroke:#1a365d,stroke-width:2px,color:#fff style B fill:#319795,stroke:#234e52,stroke-width:2px,color:#fff style C fill:#d69e2e,stroke:#744210,stroke-width:2px,color:#fff style D fill:#dd6b20,stroke:#7b341e,stroke-width:2px,color:#fff style E fill:#805ad5,stroke:#44337a,stroke-width:2px,color:#fff style F fill:#e53e3e,stroke:#742a2a,stroke-width:2px,color:#fff ``` ### 1. 遥测摄取子系统 (`ingestion/`) 一个线程安全、可扩展的流处理引擎，能够通过滑动窗口或批处理配置处理高速事件流。 * **实时摄取器 (`realtime_ingestor.py`)：** 核心摄取循环，利用并发工作线程和基于 deque 的缓冲区进行非阻塞事件处理。支持动态批处理大小和滑动时间窗口。 * **企业连接器 (`connectors.py`)：** 可插拔的生成器函数，支持与 Kafka、Elasticsearch、Splunk、REST API、WebSockets、Syslog 和 Windows 事件日志的原生集成。 **示例：Kafka 连接器集成** ``` def kafka_event_stream( topic: str, bootstrap_servers: List[str], group_id: str = "threat-intelligence-engine", ) -> Generator[Dict[str, Any], None, None]: # ... connection logic ... for message in consumer: raw_event = message.value # Normalize to standard schema yield { "campaign_id": raw_event.get("campaign_id", "KAFKA-STREAM"), "event_id": raw_event.get("event_id", str(uuid.uuid4())), "timestamp": raw_event.get("timestamp", str(datetime.now(timezone.utc))), "host": raw_event.get("host", "UNKNOWN"), "technique_id": raw_event.get("technique_id", None), "source": "kafka", "_raw": raw_event, } ``` * **Sysmon 解析器 (`sysmon_parser.py`)：** 确定性解析器，用于从原始 Sysmon 日志中推断对抗性技术。 **示例：确定性技术推断** ``` def infer_technique(event): event_id = event.get("EventID") command = str(event.get("CommandLine", "")).lower() # Process Creation (Event ID 1) if event_id == 1: # Encoded PowerShell -> Execution (T1059) if "powershell" in image and "-enc" in command: return "T1059" # Mimikatz execution -> Credential Dumping (T1003) if "mimikatz" in command: return "T1003" # Network Connection (Event ID 3) -> Exfiltration (T1041) if event_id == 3: return "T1041" return None ``` ### 2. 对抗性映射层 (`mapping/`) * **MITRE ATT&CK 映射器 (`mitre_mapper.py`)：** 通过将原始遥测和推断出的技术转化为标准化的 MITRE ATT&CK 战术（例如：执行、持久化、凭证访问），对不同的事件流进行标准化处理。 ### 3. 分析与指纹识别引擎 (`analytics/`) 负责将离散事件转化为可操作情报的计算核心。 * **指纹引擎 (`fingerprint_engine.py`)：** 生成代表对手行为的定长行为向量。计算技术频率、战术序列、主机传播范围以及专有的**活动复杂度评分**。 **示例：行为向量化** ``` # 定长向量化映射 ALL_TECHNIQUES = ["T1059", "T1547", "T1003", "T1021", "T1041"] def generate_fingerprint(mapped_logs): techniques = [e["technique_id"] for e in mapped_logs] hosts = set(e["host"] for e in mapped_logs) technique_freq = Counter(techniques) # TRUE BEHAVIORAL VECTOR (Technique-based) # Each dimension represents frequency of a specific technique vector = np.array([ technique_freq.get(t, 0) for t in ALL_TECHNIQUES ]) # Behavioral complexity metric complexity_score = len(set(techniques)) * len(hosts) # ... returns comprehensive fingerprint dict ... ``` * **聚类引擎 (`clustering_engine.py`)：** 利用 `scikit-learn` 的 K-Means 聚类对行为向量进行处理，将结构相似的活动分组，并识别威胁行为者的重叠。 **示例：基于行为向量的 K-Means 聚类** ``` from sklearn.cluster import KMeans import numpy as np def cluster_campaigns(fingerprints, n_clusters=2): # Extract behavioral vectors vectors = np.array([fp["vector"] for fp in fingerprints]) # Dynamic cluster adjustment based on distinct vectors if len(set(map(tuple, vectors))) < n_clusters: n_clusters = len(set(map(tuple, vectors))) # Execute K-Means clustering kmeans = KMeans(n_clusters=n_clusters, random_state=42, n_init=10) labels = kmeans.fit_predict(vectors) return labels ``` * **相似度引擎 (`similarity_engine.py`)：** 计算活动向量之间的余弦相似度，以量化操作差异。 **示例：量化活动差异** ``` from sklearn.metrics.pairwise import cosine_similarity import numpy as np def compute_similarity(fp1, fp2): # Reshape behavioral vectors for scikit-learn v1 = np.array(fp1["vector"]).reshape(1, -1) v2 = np.array(fp2["vector"]).reshape(1, -1) # Compute cosine similarity score (1.0 = identical, 0.0 = completely divergent) score = cosine_similarity(v1, v2)[0][0] return score ``` * **图构建器 (`graph_builder.py`)：** 使用 `networkx` 构建有向图，以可视化攻击技术在活动中的时间进展和关系。 **示例：构建攻击进展图** ``` import networkx as nx def build_attack_graph(mapped_logs): G = nx.DiGraph() # Map sequential technique transitions as directed edges for i in range(len(mapped_logs)-1): src = mapped_logs[i]["technique_id"] dst = mapped_logs[i+1]["technique_id"] G.add_edge(src, dst) return G ``` ### 4. 情报报告 (`reporting/`) * **报告生成器 (`intelligence_report.py`)：** 将分析输出综合为战略性威胁情报报告，详细说明风险级别、集群分类和行为评估。 ## ⚙️ 操作模式 引擎支持多种操作范式，可通过 `main.py` 中的 `MODE` 指令进行配置： ``` # ========================================== # 配置 (main.py) # ========================================== # "simulate" → synthetic campaigns (batch mode) # "sysmon" → real Sysmon ingestion (batch mode) # "realtime" → real-time streaming ingestion (simulated) # "kafka" → Kafka streaming # "splunk" → Splunk integration MODE = "realtime" ``` ## 🛠️ 部署与配置 ### 前置条件 根据您所需的操作模式安装必要的依赖项： ``` # 核心依赖项 pip install pandas scikit-learn networkx numpy jinja2 # Connector 依赖项（按需安装） pip install kafka-python elasticsearch splunk-sdk requests websocket-client ``` ### 企业级配置 在 `main.py` 中修改 `CONNECTOR_CONFIG` 以填入您的企业凭证。有关全面的示例，请参阅 `ENDPOINT_CONFIGURATIONS.py`。 **示例：混合云/本地部署设置** ``` CONNECTOR_CONFIG = { "splunk": { "host": "splunk.internal.local:8089", "username": "threat_intel", "password": "SecurePassword123", # Use env vars in prod "search_query": "sourcetype=sysmon index=main earliest=-30m latest=now" }, "kafka": { "bootstrap_servers": ["kafka1.cloud.local:9092", "kafka2.cloud.local:9092"], "topic": "security.alerts", "group_id": "apt_threat_intelligence" } } ``` ### 运行引擎 **1. 交互式演示 (`connector_demo.py`)** 要快速了解所有受支持的企业集成和配置，请运行交互式演示： ``` python connector_demo.py ``` 该实用程序提供了一个终端 UI，用于比较连接器的速度、规模以及特定的部署命令。 **2. 主流水线执行 (`main.py`)** 要以您配置的模式（批处理、模拟或实时连接器）运行引擎： ``` python main.py ``` ### 测试基础设施 该项目包含一套 mock 服务器，无需访问生产系统即可验证摄取流水线： ``` # 启动所有 mock 服务器 (REST、WebSocket、Splunk) python setup_infrastructure.py # 发送模拟的 Syslog 事件以测试 ingestion python send_syslog.py localhost 514 10 ``` ## 🔬 分析方法：行为向量化 该引擎摒弃了传统的特征匹配方式，转而采用**行为向量化**。 1. **提取：** 解析遥测数据以识别执行工件、网络连接和注册表修改。 2. **推断：** 将工件映射到特定的 MITRE ATT&CK 技术（例如：带混淆的 PowerShell 执行 -> T1059）。 3. **向量化：** 将技术汇总为预定义技术域内的频率分布向量。 4. **聚类：** 向量被映射到 n 维空间中。K-Means 聚类识别代表不同操作 playbook 的质心，从而实现对归因于同一威胁行为者的看似无关的活动进行关联。 ## 📊 脱敏的样本数据 为了便于测试和验证，代码库在 `data/` 目录下提供了预打包的、已脱敏的样本数据集： * **原始日志 (`data/raw_logs_*.json`)**：表示捕获来自各种主机的执行、网络连接和注册表修改信号的原始事件日志。敏感字段（如机器 IP 地址、域名和用户账户）已经过脱敏处理（例如，映射为 `HOST-1` 这样的通用结构和虚拟哈希值）。 * **活动数据 (`data/campaign_*.json`)**：被结构化为特定攻击活动的标准化遥测数据，按时间顺序映射事件并推断出 MITRE ATT&CK 技术，例如： - `T1059` (命令和脚本解释器) - `T1003` (凭证转储) - `T1547` (启动或登录自动启动执行) - `T1021` (远程服务) - `T1041` (通过 C2 通道外传) 这些数据集允许您立即运行聚类和相似度引擎，并验证输出结果。 ## ⚠️ 安全注意事项 * **凭证管理：** 切勿在 `CONNECTOR_CONFIG` 或 `ENDPOINT_CONFIGURATIONS.py` 中硬编码凭证。在生产部署中，请使用环境变量或安全的 vault 集成。 * **数据隐私：** 如果处理 PII 或敏感主机名，请确保对遥测流应用适当的匿名化或掩码处理。 *专为高级威胁 hunting 和主动对手特征描绘而开发。*

标签：Cloudflare, Kafka, K-Means聚类, MITRE ATT&CK, SonarQube插件, 威胁情报, 安全遥测, 开发者工具, 特权检测, 逆向工具