基于动态指令追踪与符号执行对抗OLLVM控制流平坦化的工程化实践

// trace_ollvm.js
const MODULE_BASE = Module.getBaseAddress("target.dll");
const MODULE_SIZE = Module.getSize("target.dll");

function should_trace(addr) {
    return addr.compare(MODULE_BASE) >= 0 && 
           addr.compare(MODULE_BASE.add(MODULE_SIZE)) < 0;
}

Interceptor.attach(Module.findExportByName("target.dll", "ExportEntry"), {
    onEnter(args) {
        console.log("[*] Triggering export, starting stalker...");
        Stalker.follow(this.threadId, {
            events: {
                block: true,   // 回调基本块
                call: false,   // 忽略call以减少噪音
                ret: false
            },
            onReceive(events) {
                const items = Stalker.parse(events, { 
                    annotate: true, 
                    stringify: false 
                });
                let trace_log = [];
                for (let item of items) {
                    if (item.kind === 'block') {
                        const start = item.start;
                        const end = item.end;
                        // 仅记录模块内且包含分支指令的块（x64末4字节判断）
                        if (should_trace(start)) {
                            // 读取块末尾的跳转目标（通过反汇编或直接读寄存器）
                            const last_insn = Instruction.parse(end.sub(4));
                            if (last_insn && last_insn.mnemonic.includes('jmp')) {
                                trace_log.push({
                                    addr: start.toString(16),
                                    target: last_insn.operands[0].value?.toString(16) || 'reg'
                                });
                            }
                        }
                    }
                }
                send({ type: 'trace', data: trace_log, tid: this.threadId });
            }
        });
    },
    onLeave(ret) {
        Stalker.unfollow(this.threadId);
        console.log("[*] Stalker stopped.");
    }
});

# collector.py
import json, sys, collections
from dataclasses import dataclass, asdict

@dataclass
class BlockRecord:
    addr: int
    target_addr: int  # 绝对目标地址
    hit_count: int = 0

addr_map = {}  # key: 块首地址, value: BlockRecord

def process_trace(raw_list):
    for item in raw_list:
        base = int(item['addr'], 16)
        tgt_str = item['target']
        if 'x' not in tgt_str: 
            continue  # 丢弃寄存器间接跳转（后续符号化处理）
        tgt = int(tgt_str, 16)
        if base not in addr_map:
            addr_map[base] = BlockRecord(base, tgt, 0)
        addr_map[base].hit_count += 1

# 保存聚合结果，用于后续筛选高频基本块（高频通常为分发器）
with open("trace_aggregated.json", "w") as f:
    json.dump([asdict(v) for v in addr_map.values()], f, indent=2)

import pefile, capstone, struct

pe = pefile.PE("target.dll")
base_addr = pe.OPTIONAL_HEADER.ImageBase
code_section = pe.sections[0]  # 假设 .text

def read_bytes(rva, size):
    offset = code_section.get_offset_from_rva(rva)
    return code_section.get_data()[offset:offset+size]

# 假设通过动态分析得知分发器 RVA 为 0x1000
dispatcher_rva = 0x1000
dispatcher_bytes = read_bytes(dispatcher_rva, 0x40)

import angr, claripy

proj = angr.Project("target.dll", auto_load_libs=False)
# 0x1000 是分发器虚拟地址 (RVA + ImageBase)
start_addr = base_addr + dispatcher_rva

# 初始化状态：设置栈指针和关键寄存器
state = proj.factory.blank_state(addr=start_addr)
state.regs.ebp = state.regs.esp + 0x100  # 简单栈帧

# 核心：将 eax 符号化（实际状态变量）
symbolic_eax = claripy.BVS("state_var", 32)
state.regs.eax = symbolic_eax

# 添加内存约束：保证跳转表可读（防止符号执行跑飞）
# 跳转表通常在分发器附近，通过动态 trace 可获取
jump_table_addr = base_addr + 0x2000
state.memory.store(jump_table_addr, claripy.BVV(0, 64*8), 64*8)  # 占位

# 执行至返回/间接跳转
simgr = proj.factory.simulation_manager(state)
simgr.explore(find=lambda s: s.regs.pc == jump_table_addr + 8)  # 简化：命中跳转表

# 求解真实跳转索引
if simgr.found:
    found_state = simgr.found[0]
    solution = found_state.solver.eval(symbolic_eax)
    real_target_rva = (solution & 0xFFFF) * 8 + (jump_table_addr - base_addr)  
    # 实际需解析表项，这里简化为索引计算
    print(f"[+] Symbolic eax resolved to: {hex(solution)}, Target RVA: {hex(real_target_rva)}")

def valid_path(state):
    try:
        val = state.solver.eval(state.regs.eax)
        return 0 <= val < 64  # 跳转表最多64项
    except:
        return False
simgr.explore(find=valid_path, avoid=...)

import lief
from capstone import *

binary = lief.parse("target.dll")
text_section = binary.get_section(".text")
text_content = bytearray(text_section.content)

# 假设通过前两步得到修补映射表: {dispatcher_rva: target_rva}
patch_map = {0x1000: 0x1150, 0x1020: 0x1200}  

md = Cs(CS_ARCH_X86, CS_MODE_64)
md.detail = True

for old_rva, new_rva in patch_map.items():
    offset = text_section.offset + old_rva  # 文件偏移
    
    # 1. 验证是否是 mov eax, imm; jmp [base] 模式（前4字节检查）
    insns = list(md.disasm(text_content[offset:offset+32], old_rva))
    if len(insns) < 2 or insns[0].mnemonic != 'mov' or insns[1].mnemonic != 'jmp':
        print(f"[-] Pattern mismatch at {hex(old_rva)}, skip.")
        continue
    
    # 2. 构建绝对跳转机器码: FF 25 00 00 00 00 [target 8 bytes]
    abs_jmp = bytes.fromhex("FF 25 00 00 00 00") + struct.pack("<Q", base_addr + new_rva)
    
    # 3. 填充 NOP 保持节区大小不变（原始片段通常 > 14字节）
    nop_padding = b'\x90' * (0x30 - len(abs_jmp))  # 原片段大小0x30
    new_bytes = abs_jmp + nop_padding
    
    # 4. 写回
    text_content[offset:offset+len(new_bytes)] = new_bytes

# 更新二进制
text_section.content = list(text_content)
binary.write("target_patched.dll")

for reloc in binary.relocations:
    for entry in reloc.entries:
        if entry.address >= old_rva and entry.address < old_rva + 0x30:
            entry.type = lief.PE.RELOCATIONS_BASE_TYPES.ABSOLUTE