第17章：恶意软件分析与威胁情报

恶意行为模式示例：

1. 进程注入链：
   OpenProcess → VirtualAllocEx → WriteProcessMemory → CreateRemoteThread

2. 持久化机制：
   RegOpenKeyEx → RegSetValueEx (针对Run键)
   CreateService → StartService

3. 反沙箱检测：
   GetTickCount → Sleep → GetTickCount (时间检测)
   IsDebuggerPresent → CheckRemoteDebuggerPresent

特征向量 F = {f₁, f₂, ..., fₙ}
其中：

- f₁: 熵值 H(X) = -Σ p(xᵢ)log₂p(xᵢ)
- f₂: 控制流复杂度 CC = E - N + 2P
- f₃: 字符串混淆度 O = |encrypted_strings| / |total_strings|

输入序列 S = [syscall₁, syscall₂, ..., syscallₙ]
输出：行为标签 L ∈ {文件加密, 数据窃取, 横向移动, ...}

转换函数：
P(L|S) = Π P(lᵢ|s₁, ..., sᵢ, θ)

   进程内存空间
   ┌─────────────┐
   │   Stack     │ ← 异常的栈操作
   ├─────────────┤
   │    Heap     │ ← 堆喷射检测
   ├─────────────┤
   │    Code     │ ← 自修改代码
   ├─────────────┤
   │  Injected   │ ← 注入代码区域
   └─────────────┘

低层事件序列 → LLM → 高层行为描述

例如：
[CreateFile, WriteFile, DeleteFile] + 文件路径模式
→ "勒索软件正在加密用户文档"

[WSASocket, connect, send] + 网络流量特征
→ "建立C2通道并外传敏感数据"

Context-Aware Detection Score:
Score(behavior) = α·Static_Score + β·Dynamic_Score + γ·Context_Score

其中Context_Score由LLM计算：
Context_Score = LLM(current_behavior, historical_behaviors, environment_info)

相似度计算：
Sim(AST₁, AST₂) = |Common_Subtrees| / max(|AST₁|, |AST₂|)

优化的树编辑距离：
TED(T₁, T₂) = min{cost(edit_sequence)}

代码片段嵌入：
embed(code) = LLM_encoder(preprocess(code))

余弦相似度：
cos_sim(code₁, code₂) = embed(code₁)·embed(code₂) / (||embed(code₁)|| × ||embed(code₂)||)

距离矩阵 D：
D[i,j] = distance(sample_i, sample_j)

聚类过程：

1. 初始化：每个样本为独立簇
2. 迭代合并：
   - 找到最近的两个簇
   - 合并成新簇
   - 更新距离矩阵
3. 直到达到预设簇数或距离阈值

核心思想：

- ε-邻域：N_ε(p) = {q ∈ D | dist(p,q) ≤ ε}
- 核心点：|N_ε(p)| ≥ MinPts
- 密度可达：存在点链 p₁,...,pₙ，每个点都在前一个点的ε-邻域内

变种检测：
同一家族的变种往往形成密集区域，新变种会被分配到最近的密集簇

演化图 G = (V, E)：

- V：恶意软件样本节点
- E：演化关系边

演化距离：
d_evolution(v₁, v₂) = Σ(feature_changes) × time_weight

时间衰减因子：
weight(t) = e^(-λt)，其中λ为衰减率

变异类型识别：

1. 死代码插入：识别无效的指令序列
2. 寄存器重分配：检测功能等价但寄存器不同的代码
3. 控制流变换：识别等价的控制流结构
4. API替换：识别功能相似的API调用替换

变异检测函数：
is_variant(code₁, code₂) = LLM_classify(diff(normalize(code₁), normalize(code₂)))

流量特征向量：
F_traffic = {
    packet_size_distribution,
    inter_arrival_time,
    protocol_headers,
    payload_entropy,
    periodic_patterns
}

周期性检测：
使用傅里叶变换识别beacon间隔：
F(ω) = Σ traffic(t) × e^(-iωt)

TLS指纹识别：
JA3 = MD5(TLSVersion,Ciphers,Extensions,Curves,PointFormats)

时序侧信道：

- 数据包大小分布
- 传输时间间隔
- 突发模式

协议状态机 M = (S, Σ, δ, s₀, F)：

- S：状态集合
- Σ：消息类型集合
- δ：转换函数 S × Σ → S
- s₀：初始状态
- F：终止状态集合

状态推断算法：

1. 收集流量样本
2. 聚类相似消息
3. 构建状态转换图
4. 最小化状态机

使用n-gram分析识别字段边界：
P(boundary|context) = count(boundary_ngram) / count(all_ngrams)

LLM增强的字段语义推断：
field_type = LLM_infer(field_content, protocol_context)

命令识别模型：
Input: 二进制数据流 + 上下文
Output: 命令类型及参数

示例映射：
0x01 0x00 0x04 → "CONNECT command with 4-byte payload"
0x02 0x00 0x00 → "HEARTBEAT command"
0x03 [data] → "EXECUTE command with embedded payload"

     原始流量
         ↓
    [预处理层]
    去噪、对齐
         ↓
    [特征提取]
    统计特征、序列特征
         ↓
    [LLM分析]
    语义理解、模式识别
         ↓
    [协议重建]
    状态机、消息格式
         ↓
    协议规范文档

1. 侦察（Reconnaissance）
   - 信息收集指标：DNS查询、WHOIS查询、社交媒体扫描

2. 武器化（Weaponization）
   - 载荷特征：exploit + backdoor组合

3. 投递（Delivery）
   - 投递向量：钓鱼邮件、水坑攻击、供应链攻击

4. 利用（Exploitation）
   - 漏洞利用：CVE映射、0day检测

5. 安装（Installation）
   - 持久化机制：注册表、计划任务、服务

6. 命令控制（C2）
   - 通信特征：周期性beacon、加密隧道

7. 目标达成（Actions on Objectives）
   - 最终行为：数据外传、破坏、勒索

马尔可夫链模型：
P(Stage_i+1 | Stage_i) = transition_probability

检测置信度：
Confidence(attack) = Π P(evidence_i | stage_i)

ATT&CK矩阵映射：
┌─────────────┬──────────┬──────────┬──────────┐
│   Tactic    │ Tech_1   │ Tech_2   │ Tech_3   │
├─────────────┼──────────┼──────────┼──────────┤
│Initial Access│ T1566   │ T1190   │ T1133   │
│Execution    │ T1059   │ T1106   │ T1053   │
│Persistence  │ T1547   │ T1543   │ T1546   │
│Priv Escal   │ T1055   │ T1548   │ T1134   │
└─────────────┴──────────┴──────────┴──────────┘

技术检测置信度：
Score(technique) = Σ weight_i × indicator_i

输入：系统事件日志 + 网络流量
处理：

1. 事件标准化：normalize(raw_event) → structured_event
2. 上下文聚合：aggregate(events, time_window) → context
3. TTP推断：LLM(context) → {tactics, techniques, confidence}

输出：ATT&CK技术ID及置信度

时间窗口关联：
related(e1, e2) = {
    True if |timestamp(e1) - timestamp(e2)| < θ 
         AND similarity(e1, e2) > σ
    False otherwise
}

因果关系推断：
causality_score = P(e2|e1) × temporal_proximity × semantic_similarity

攻击图 AG = (N, E, λ)：

- N：系统状态节点
- E：攻击行为边
- λ：似然度函数

路径概率：
P(path) = Π P(edge_i) × P(node_i)

关键路径识别：
critical_path = argmax_path P(path) × impact(target)

CFG定义：
G = (B, E, φ)

- B：基本块集合
- E：控制流边集合
- φ：基本块特征函数

基本块特征向量：
φ(b) = [
    指令类型分布,
    操作数特征,
    API调用特征,
    语义嵌入
]

节点更新规则：
h_i^(l+1) = σ(W^(l) · AGG({h_j^(l) : j ∈ N(i)} ∪ {h_i^(l)}))

其中：

- h_i^(l)：第l层节点i的隐藏状态
- N(i)：节点i的邻居集合
- AGG：聚合函数（mean、max、sum）
- W^(l)：第l层的权重矩阵
- σ：激活函数

图级表示：
h_G = READOUT({h_i^(L) : i ∈ G})

图注意力网络（GAT）：
α_ij = softmax_j(LeakyReLU(a^T[Wh_i || Wh_j]))

h_i' = σ(Σ_{j∈N(i)} α_ij · W·h_j)

多头注意力：
h_i' = ||_{k=1}^K σ(Σ_{j∈N(i)} α_ij^k · W^k·h_j)

函数调用图 FCG = (F, C, ψ)：

- F：函数节点集合
- C：调用边集合
- ψ：函数特征提取器

函数特征：
ψ(f) = [
    循环复杂度,
    参数个数,
    局部变量数,
    调用深度,
    返回值类型,
    CFG嵌入
]

两级架构：
Level 1: CFG-GNN
  输入：函数内控制流图
  输出：函数嵌入 e_f

Level 2: FCG-GNN
  输入：函数调用图 + 函数嵌入
  输出：程序嵌入 e_p

组合公式：
e_p = FCG-GNN(G_fcg, {e_f : f ∈ Functions})

跨图注意力：
attention(g1, g2) = softmax(similarity(h_g1, h_g2))

匹配分数：
match_score = σ(MLP([h_g1; h_g2; h_g1 - h_g2; h_g1 ⊙ h_g2]))

其中⊙表示逐元素乘积

通用特征提取：

1. 语义操作序列：[LOAD, ADD, STORE, BRANCH, ...]
2. 数据流模式：def-use链
3. 循环结构：loop invariants
4. 调用约定：参数传递模式

归一化表示：
normalize(instruction) → abstract_operation

域适应模型：
Source: x86架构样本
Target: ARM架构样本

对抗训练：
L_total = L_classify + λ·L_domain

其中：
L_classify：分类损失
L_domain：域判别器损失（最大化混淆）

相似度融合：
Sim_final = Σ w_i · Sim_i

模态包括：

- Sim_cfg：控制流相似度
- Sim_api：API调用相似度
- Sim_string：字符串相似度
- Sim_behavior：行为相似度

权重学习：
w = softmax(MLP([features]))

聚类流程：

1. 样本预处理：反汇编、CFG提取
2. 图嵌入生成：GNN编码
3. 相似度计算：余弦距离
4. 层次聚类：Ward链接
5. 家族标注：专家验证

评估指标：

- 同质性：V_homogeneity
- 完整性：V_completeness
- V-measure：2 × (homogeneity × completeness) / (homogeneity + completeness)

异常检测模型：
score(sample) = min_family distance(sample, family_center)

如果 score > threshold：
  可能的零日变种

自适应阈值：
threshold = μ + k·σ
其中k根据false positive rate动态调整

CreateToolhelp32Snapshot → Process32First → Process32Next → 
OpenProcess → VirtualAllocEx → WriteProcessMemory → CreateRemoteThread

def comprehensive_analysis(sample):
    static_features = extract_static(sample)
    dynamic_features = sandbox_execution(sample)
    semantic_features = llm_analysis(sample)

    return combine_features(static_features, 
                          dynamic_features, 
                          semantic_features)

def temporal_correlation(events, time_window=3600):
    event_chains = []
    for event in events:
        related = find_related_events(event, time_window)
        if len(related) > threshold:
            event_chains.append(build_chain(event, related))
    return merge_overlapping_chains(event_chains)

# 1. 重采样
balanced_data = SMOTE(original_data)

# 2. 类别权重
class_weights = compute_class_weight('balanced', 
                                    classes, 
                                    y_train)

# 3. Focal Loss
focal_loss = -α * (1-p)^γ * log(p)

def adversarial_training(model, data):
    for epoch in range(epochs):
        # 生成对抗样本
        adv_samples = generate_adversarial(data, model)

        # 混合训练
        mixed_data = combine(data, adv_samples)

        # 鲁棒性正则化
        loss = classification_loss + λ * robustness_loss

        model.train(mixed_data, loss)