1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14

; ARM64 可执行花指令示例
nop                     ; 空操作
mov x0, x0              ; 自身赋值
add x1, x1, #0          ; 加0操作
eor x2, x2, #0          ; 异或0操作
and x3, x3, #-1         ; 与全1操作
orr x4, x4, #0          ; 或0操作

; 更复杂的等价操作
push {x0}               ; 压栈
pop {x0}                ; 出栈（恢复原值）

sub sp, sp, #16         ; 分配栈空间
add sp, sp, #16         ; 释放栈空间（抵消）

 1
 2
 3
 4
 5
 6
 7
 8
 9
10

; ARM64 不可达花指令示例
    cmp x0, x0          ; 比较相同值
    b.ne junk_code      ; 永假条件跳转
    ; ... 正常代码 ...
    b continue
junk_code:
    .byte 0xFF, 0xFF, 0xFF, 0xFF  ; 垃圾数据
    .byte 0x00, 0x00, 0x00, 0x00
continue:
    ; ... 继续正常执行 ...

1
2
3
4
5
6

; x86 数据混合示例
    jmp skip_data       ; 跳过数据区
    .byte 0xE8          ; 看起来像 call 指令
    .byte 0x90, 0x90    ; 填充数据
skip_data:
    mov eax, ebx        ; 正常代码

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19

// C++ 编译时花指令生成
#define JUNK_CODE_1() \
    __asm__ volatile ( \
        "cmp x0, x0\n\t" \
        "b.ne 1f\n\t" \
        "b 2f\n\t" \
        "1:\n\t" \
        ".byte 0xFF, 0xFF, 0xFF, 0xFF\n\t" \
        "2:\n\t" \
    )

// 使用示例
void protected_function() {
    JUNK_CODE_1();
    // 真实代码
    int result = calculate_key();
    JUNK_CODE_1();
    return result;
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23

// 不透明谓词：看似条件分支，实际永远为真/假
int opaque_true() {
    int x = rand();
    // (x * x) % 2 == 0 永远为真（平方数的奇偶性）
    return ((x * x) % 2 == 0);
}

int opaque_false() {
    int x = rand();
    // (x * (x + 1)) % 2 != 0 永远为假
    return ((x * (x + 1)) % 2 != 0);
}

void obfuscated_function() {
    if (opaque_true()) {
        // 真实代码路径
        real_logic();
    } else {
        // 垃圾代码，永远不会执行
        garbage_code_1();
        garbage_code_2();
    }
}

1
2
3
4
5

; x86 多字节指令欺骗
    jmp short skip + 1  ; 跳到下一条指令的中间
skip:
    .byte 0xE8          ; 看起来是 CALL 指令开始
    mov eax, 1          ; 实际执行的指令

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96

# IDA Python 脚本：识别可疑的花指令模式
import idautils
import idc
import idaapi

class JunkCodeDetector:
    def __init__(self):
        self.junk_patterns = []

    def detect_nop_sleds(self, func_ea):
        """检测 NOP 滑板"""
        nop_count = 0
        for head in idautils.FuncItems(func_ea):
            mnem = idc.print_insn_mnem(head)
            if mnem == "NOP" or mnem == "nop":
                nop_count += 1
            else:
                if nop_count > 3:
                    print(f"[!] NOP sled detected at 0x{head-nop_count:x}, count: {nop_count}")
                    self.junk_patterns.append((head - nop_count, nop_count, "NOP_SLED"))
                nop_count = 0
        return self.junk_patterns

    def detect_dead_code(self, func_ea):
        """检测死代码（永远不会执行的代码）"""
        func = idaapi.get_func(func_ea)
        if not func:
            return []

        # 获取函数的控制流图
        flowchart = idaapi.FlowChart(func)
        reachable_blocks = set()

        # BFS 遍历可达基本块
        queue = [flowchart[0].start_ea]
        while queue:
            block_ea = queue.pop(0)
            if block_ea in reachable_blocks:
                continue
            reachable_blocks.add(block_ea)

            for block in flowchart:
                if block.start_ea == block_ea:
                    for succ in block.succs():
                        queue.append(succ.start_ea)
                    break

        # 找出不可达的基本块
        dead_blocks = []
        for block in flowchart:
            if block.start_ea not in reachable_blocks:
                print(f"[!] Dead code block at 0x{block.start_ea:x}")
                dead_blocks.append(block.start_ea)

        return dead_blocks

    def detect_opaque_predicates(self, func_ea):
        """检测不透明谓词"""
        suspicious_patterns = []

        for head in idautils.FuncItems(func_ea):
            mnem = idc.print_insn_mnem(head)

            # 检测 cmp reg, reg 后跟条件跳转
            if mnem.lower() == "cmp":
                op1 = idc.print_operand(head, 0)
                op2 = idc.print_operand(head, 1)

                if op1 == op2:
                    next_insn = idc.next_head(head)
                    next_mnem = idc.print_insn_mnem(next_insn)

                    if next_mnem.lower().startswith("j") or next_mnem.lower().startswith("b"):
                        print(f"[!] Opaque predicate at 0x{head:x}: {mnem} {op1}, {op2}")
                        suspicious_patterns.append((head, "OPAQUE_PREDICATE"))

        return suspicious_patterns

def analyze_function_junk_code(func_ea):
    """分析函数中的花指令"""
    detector = JunkCodeDetector()

    print(f"[*] Analyzing function at 0x{func_ea:x}")

    nop_sleds = detector.detect_nop_sleds(func_ea)
    dead_blocks = detector.detect_dead_code(func_ea)
    opaque_preds = detector.detect_opaque_predicates(func_ea)

    return {
        'nop_sleds': nop_sleds,
        'dead_blocks': dead_blocks,
        'opaque_predicates': opaque_preds
    }

# 使用示例
# results = analyze_function_junk_code(idc.here())

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38

# 花指令模式签名库
ARM64_JUNK_PATTERNS = [
    # (pattern_bytes, mask, description)
    (b'\x1f\x00\x00\xeb', b'\xff\xff\xff\xff', 'cmp x0, x0'),
    (b'\x00\x00\x00\x14', b'\x00\x00\x00\x1f', 'unconditional branch to next'),
    (b'\xe0\x03\x00\xaa', b'\xff\xff\xff\xff', 'mov x0, x0'),
]

X86_JUNK_PATTERNS = [
    (b'\x90', b'\xff', 'nop'),
    (b'\x89\xc0', b'\xff\xff', 'mov eax, eax'),
    (b'\x31\xc0\x31\xc0', b'\xff\xff\xff\xff', 'xor eax, eax; xor eax, eax'),
    (b'\xeb\x00', b'\xff\xff', 'jmp $+2 (next instruction)'),
]

def scan_junk_patterns(binary_data, patterns, base_addr=0):
    """扫描二进制数据中的花指令模式"""
    findings = []

    for offset in range(len(binary_data)):
        for pattern, mask, desc in patterns:
            if offset + len(pattern) > len(binary_data):
                continue

            match = True
            for i in range(len(pattern)):
                if (binary_data[offset + i] & mask[i]) != (pattern[i] & mask[i]):
                    match = False
                    break

            if match:
                findings.append({
                    'address': base_addr + offset,
                    'pattern': pattern.hex(),
                    'description': desc
                })

    return findings

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88

// Frida 脚本：跟踪实际执行的指令，区分花指令
function trace_real_execution(module_name, func_offset, trace_count) {
    var module = Process.findModuleByName(module_name);
    if (!module) {
        console.log("Module not found: " + module_name);
        return;
    }

    var func_addr = module.base.add(func_offset);
    var executed_addresses = new Set();
    var instruction_count = 0;

    Stalker.follow(Process.getCurrentThreadId(), {
        transform: function(iterator) {
            var instruction = iterator.next();

            while (instruction !== null) {
                var addr = instruction.address;

                // 检查是否在目标模块范围内
                if (addr.compare(module.base) >= 0 &&
                    addr.compare(module.base.add(module.size)) < 0) {

                    iterator.putCallout(function(context) {
                        var pc = context.pc;
                        if (!executed_addresses.has(pc.toString())) {
                            executed_addresses.add(pc.toString());

                            var insn = Instruction.parse(pc);
                            console.log("[EXEC] " + pc + ": " + insn.mnemonic + " " + insn.opStr);

                            instruction_count++;
                            if (instruction_count >= trace_count) {
                                Stalker.unfollow();
                            }
                        }
                    });
                }

                iterator.keep();
                instruction = iterator.next();
            }
        }
    });

    // 调用目标函数
    var target_func = new NativeFunction(func_addr, 'void', []);
    target_func();

    Stalker.unfollow();
    Stalker.flush();

    console.log("\n[*] Total unique instructions executed: " + executed_addresses.size);
    return Array.from(executed_addresses);
}

// 对比静态反汇编结果与动态执行结果
function identify_junk_by_execution(module_name, func_offset) {
    var module = Process.findModuleByName(module_name);
    var func_addr = module.base.add(func_offset);

    // 获取静态反汇编的所有地址
    var static_addresses = [];
    var current = func_addr;
    for (var i = 0; i < 100; i++) { // 假设函数不超过100条指令
        try {
            var insn = Instruction.parse(current);
            static_addresses.push(current.toString());
            current = current.add(insn.size);
        } catch (e) {
            break;
        }
    }

    // 获取动态执行的地址
    var executed = trace_real_execution(module_name, func_offset, 1000);

    // 找出从未执行的指令（潜在花指令）
    var junk_candidates = static_addresses.filter(addr => !executed.includes(addr));

    console.log("\n[*] Potential junk code (never executed):");
    junk_candidates.forEach(function(addr) {
        var insn = Instruction.parse(ptr(addr));
        console.log("  " + addr + ": " + insn.mnemonic + " " + insn.opStr);
    });

    return junk_candidates;
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96

from unicorn import *
from unicorn.arm64_const import *
from capstone import *
import struct

class JunkCodeRemover:
    def __init__(self, binary_data, base_addr=0x10000):
        self.binary = binary_data
        self.base_addr = base_addr
        self.executed_addresses = set()
        self.md = Cs(CS_ARCH_ARM64, CS_MODE_ARM)

    def hook_code(self, uc, address, size, user_data):
        """记录执行的每条指令"""
        self.executed_addresses.add(address)

        # 打印执行的指令
        code = uc.mem_read(address, size)
        for insn in self.md.disasm(bytes(code), address):
            print(f"[EXEC] 0x{insn.address:x}: {insn.mnemonic} {insn.op_str}")

    def emulate_function(self, func_offset, max_instructions=1000):
        """模拟执行函数，收集执行路径"""
        try:
            # 初始化 Unicorn 引擎 (ARM64)
            mu = Uc(UC_ARCH_ARM64, UC_MODE_ARM)

            # 映射内存
            mu.mem_map(self.base_addr, 2 * 1024 * 1024)  # 2MB
            mu.mem_write(self.base_addr, self.binary)

            # 设置栈
            stack_addr = self.base_addr + 0x100000
            mu.reg_write(UC_ARM64_REG_SP, stack_addr)

            # 设置代码 Hook
            mu.hook_add(UC_HOOK_CODE, self.hook_code)

            # 开始模拟
            func_addr = self.base_addr + func_offset
            mu.emu_start(func_addr, func_addr + 0x10000, count=max_instructions)

        except UcError as e:
            print(f"Emulation error: {e}")

        return self.executed_addresses

    def identify_junk_code(self, func_offset, func_size):
        """识别函数中的花指令"""
        # 静态反汇编获取所有指令地址
        static_addresses = set()
        func_binary = self.binary[func_offset:func_offset + func_size]

        for insn in self.md.disasm(func_binary, self.base_addr + func_offset):
            static_addresses.add(insn.address)

        # 模拟执行获取实际执行的指令
        self.emulate_function(func_offset)

        # 差集即为花指令
        junk_addresses = static_addresses - self.executed_addresses

        print(f"\n[*] Static instructions: {len(static_addresses)}")
        print(f"[*] Executed instructions: {len(self.executed_addresses)}")
        print(f"[*] Junk instructions: {len(junk_addresses)}")

        # 输出花指令详情
        print("\n[*] Junk code details:")
        for addr in sorted(junk_addresses):
            offset = addr - self.base_addr
            code = self.binary[offset:offset+4]
            for insn in self.md.disasm(code, addr):
                print(f"  0x{insn.address:x}: {insn.mnemonic} {insn.op_str}")

        return junk_addresses

    def patch_junk_code(self, junk_addresses, output_path):
        """将花指令替换为 NOP"""
        patched_binary = bytearray(self.binary)

        for addr in junk_addresses:
            offset = addr - self.base_addr
            # ARM64 NOP = 0x1F 0x20 0x03 0xD5
            patched_binary[offset:offset+4] = b'\x1f\x20\x03\xd5'

        with open(output_path, 'wb') as f:
            f.write(patched_binary)

        print(f"[*] Patched binary saved to {output_path}")

# 使用示例
# with open('target.so', 'rb') as f:
#     binary = f.read()
# remover = JunkCodeRemover(binary)
# junk = remover.identify_junk_code(func_offset=0x1234, func_size=0x200)
# remover.patch_junk_code(junk, 'target_cleaned.so')

  1
  2
  3
  4
  5
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137

# IDA Pro 花指令自动清除插件
import idaapi
import idc
import idautils

class JunkCodeCleaner(idaapi.plugin_t):
    flags = idaapi.PLUGIN_UNL
    comment = "Automatic Junk Code Cleaner"
    help = "Clean junk code patterns"
    wanted_name = "Junk Cleaner"
    wanted_hotkey = "Ctrl-Shift-J"

    def init(self):
        return idaapi.PLUGIN_OK

    def run(self, arg):
        func_ea = idc.get_screen_ea()
        func = idaapi.get_func(func_ea)

        if not func:
            print("[!] Not inside a function")
            return

        self.clean_function(func.start_ea)

    def clean_function(self, func_ea):
        """清理函数中的花指令"""
        cleaned_count = 0

        # 1. 清理 NOP 滑板
        cleaned_count += self.remove_nop_sleds(func_ea)

        # 2. 清理不透明谓词
        cleaned_count += self.remove_opaque_predicates(func_ea)

        # 3. 清理死代码
        cleaned_count += self.remove_dead_code(func_ea)

        # 4. 重新分析函数
        idaapi.reanalyze_callers(func_ea, True)

        print(f"[*] Cleaned {cleaned_count} junk patterns in function 0x{func_ea:x}")

    def remove_nop_sleds(self, func_ea):
        """移除 NOP 滑板"""
        removed = 0
        nop_start = None
        nop_count = 0

        for head in idautils.FuncItems(func_ea):
            mnem = idc.print_insn_mnem(head).lower()

            if mnem == "nop":
                if nop_start is None:
                    nop_start = head
                nop_count += 1
            else:
                if nop_count > 3:
                    # 将多余的 NOP 转换为数据
                    for i in range(nop_count - 1):
                        idc.del_items(nop_start + i)
                    removed += nop_count - 1
                nop_start = None
                nop_count = 0

        return removed

    def remove_opaque_predicates(self, func_ea):
        """移除不透明谓词"""
        removed = 0

        for head in idautils.FuncItems(func_ea):
            mnem = idc.print_insn_mnem(head).lower()

            if mnem == "cmp":
                op1 = idc.print_operand(head, 0)
                op2 = idc.print_operand(head, 1)

                if op1 == op2:
                    # cmp reg, reg 总是相等
                    next_insn = idc.next_head(head)
                    next_mnem = idc.print_insn_mnem(next_insn).lower()

                    # 根据条件类型 patch
                    if next_mnem in ["jne", "jnz", "b.ne"]:
                        # 条件永假，删除跳转目标的代码块
                        target = idc.get_operand_value(next_insn, 0)
                        self.mark_as_junk(target)
                        removed += 1
                    elif next_mnem in ["je", "jz", "b.eq"]:
                        # 条件永真，可以转换为无条件跳转
                        idc.patch_byte(next_insn, 0xEB)  # jmp short
                        removed += 1

        return removed

    def remove_dead_code(self, func_ea):
        """移除死代码"""
        func = idaapi.get_func(func_ea)
        if not func:
            return 0

        flowchart = idaapi.FlowChart(func)
        reachable = set()

        # BFS 找可达块
        queue = [flowchart[0].start_ea]
        while queue:
            block_ea = queue.pop(0)
            if block_ea in reachable:
                continue
            reachable.add(block_ea)

            for block in flowchart:
                if block.start_ea == block_ea:
                    for succ in block.succs():
                        queue.append(succ.start_ea)

        # 标记不可达块为数据
        removed = 0
        for block in flowchart:
            if block.start_ea not in reachable:
                for head in range(block.start_ea, block.end_ea):
                    idc.del_items(head)
                    removed += 1

        return removed

    def mark_as_junk(self, address):
        """将地址标记为花指令"""
        idc.set_cmt(address, "JUNK CODE - Opaque predicate target", 0)

    def term(self):
        pass

def PLUGIN_ENTRY():
    return JunkCodeCleaner()

1
2
3
4
5
6
7
8

void original_func(int x) {
    if (x > 10) {
        func_a();
    } else {
        func_b();
    }
    func_c();
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33

void flattened_func(int x) {
    int state = 0;  // 状态变量

    while (1) {
        switch (state) {
            case 0:  // 入口
                if (x > 10) {
                    state = 1;  // 去 func_a
                } else {
                    state = 2;  // 去 func_b
                }
                break;

            case 1:  // func_a 块
                func_a();
                state = 3;  // 去 func_c
                break;

            case 2:  // func_b 块
                func_b();
                state = 3;  // 去 func_c
                break;

            case 3:  // func_c 块
                func_c();
                state = 4;  // 退出
                break;

            case 4:  // 出口
                return;
        }
    }
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35

; ARM64 控制流平坦化特征
_flattened_func:
    ; 初始化状态变量
    mov     w8, #0              ; state = 0

.Lmain_loop:
    ; 主分发器 switch
    cmp     w8, #4
    b.hi    .Ldefault

    adrp    x9, .Ljump_table
    add     x9, x9, :lo12:.Ljump_table
    ldr     x9, [x9, x8, lsl #3]
    br      x9

.Lcase_0:
    ; 入口块
    cmp     w0, #10
    csel    w8, w8, w8, gt      ; 条件更新状态
    b       .Lmain_loop

.Lcase_1:
    ; func_a 块
    bl      func_a
    mov     w8, #3
    b       .Lmain_loop

; ... 其他 case ...

.Ljump_table:
    .quad   .Lcase_0
    .quad   .Lcase_1
    .quad   .Lcase_2
    .quad   .Lcase_3
    .quad   .Lcase_4

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16

void bcf_func(int x) {
    // 不透明谓词（永真）
    int opaque = (x * x >= 0);  // 平方数永远非负

    if (opaque) {
        real_logic();      // 真实代码
    } else {
        fake_logic_1();    // 永远不执行
        fake_logic_2();
    }

    // 另一个不透明谓词（永假）
    if ((x * (x + 1)) % 2 != 0) {  // 连续整数乘积永远是偶数
        garbage_code();     // 永远不执行
    }
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16

// 原始：a + b
// 替换 1：a - (-b)
// 替换 2：(a ^ b) + 2 * (a & b)
// 替换 3：2 * (a | b) - (a ^ b)

// 原始：a - b
// 替换：a + (~b + 1)
// 替换：(a ^ b) - 2 * (~a & b)

// 原始：a ^ b
// 替换：(~a & b) | (a & ~b)
// 替换：(a | b) & (~a | ~b)

// 原始：a & b
// 替换：(~(~a | ~b))
// 替换：((a ^ b) ^ (a | b))

1
2
3
4
5
6

; 原始: ADD X0, X1, X2  (x0 = x1 + x2)
; 替换后: 使用 XOR 和 AND 的组合
eor     x8, x1, x2      ; x8 = x1 ^ x2
and     x9, x1, x2      ; x9 = x1 & x2
lsl     x9, x9, #1      ; x9 = (x1 & x2) << 1 = 2*(x1&x2)
add     x0, x8, x9      ; x0 = (x1^x2) + 2*(x1&x2) = x1 + x2

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20

// OLLVM 字符串加密示例
// 原始字符串在编译时被加密，运行时解密

// 编译后的加密数据
static unsigned char encrypted_str[] = {
    0x8a, 0x45, 0x9c, 0x3f, 0x2d, 0x77, 0x4e, 0x1b
};

// 解密函数（通常内联或混淆）
char* decrypt_string() {
    static char decrypted[64];
    unsigned char key = 0xAA;

    for (int i = 0; i < sizeof(encrypted_str); i++) {
        decrypted[i] = encrypted_str[i] ^ key;
        key = rotate_left(key, 3);  // 密钥轮换
    }

    return decrypted;
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97

# 使用 angr 进行符号执行去平坦化
import angr
import claripy
from collections import defaultdict

class CFGDeflattener:
    def __init__(self, binary_path, func_addr):
        self.proj = angr.Project(binary_path, auto_load_libs=False)
        self.func_addr = func_addr
        self.state_var = None
        self.real_cfg = defaultdict(list)

    def identify_dispatcher(self):
        """识别主分发器"""
        cfg = self.proj.analyses.CFGFast()
        func = cfg.functions.get(self.func_addr)

        if not func:
            print(f"Function not found at 0x{self.func_addr:x}")
            return None

        # 寻找入度最高的基本块（可能是分发器）
        block_in_degree = defaultdict(int)
        for block in func.blocks:
            for succ in block.vex.constant_jump_targets:
                block_in_degree[succ] += 1

        dispatcher = max(block_in_degree, key=block_in_degree.get)
        print(f"[*] Identified dispatcher at 0x{dispatcher:x}")
        return dispatcher

    def symbolic_execution(self, start_addr, max_steps=1000):
        """符号执行追踪状态转移"""
        state = self.proj.factory.blank_state(addr=start_addr)

        # 创建符号状态变量
        state_sym = claripy.BVS('state', 32)

        transitions = []
        visited = set()

        simgr = self.proj.factory.simulation_manager(state)

        step_count = 0
        while simgr.active and step_count < max_steps:
            simgr.step()
            step_count += 1

            for s in simgr.active:
                current_block = s.addr

                if current_block in visited:
                    continue
                visited.add(current_block)

                # 记录状态转移
                if hasattr(s, 'history') and len(s.history.bbl_addrs) > 1:
                    prev_block = s.history.bbl_addrs[-2]
                    transitions.append((prev_block, current_block))

        return transitions

    def recover_cfg(self):
        """恢复原始控制流图"""
        dispatcher = self.identify_dispatcher()
        if not dispatcher:
            return None

        # 符号执行获取转移关系
        transitions = self.symbolic_execution(self.func_addr)

        # 过滤掉通过分发器的边
        real_transitions = []
        for src, dst in transitions:
            if src != dispatcher and dst != dispatcher:
                real_transitions.append((src, dst))

        print("\n[*] Recovered CFG edges:")
        for src, dst in real_transitions:
            print(f"  0x{src:x} -> 0x{dst:x}")

        return real_transitions

    def generate_patched_binary(self, output_path):
        """生成修复后的二进制"""
        real_cfg = self.recover_cfg()

        # 这里需要实现实际的 patch 逻辑
        # 1. 移除分发器块
        # 2. 将状态赋值替换为直接跳转
        # 3. 重新链接基本块

        print(f"[*] Patched binary saved to {output_path}")

# 使用示例
# deflattener = CFGDeflattener("target.so", 0x1234)
# deflattener.generate_patched_binary("target_deflattened.so")

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73

# IDA Pro Microcode API 去平坦化
import ida_hexrays
import ida_funcs
import idautils

class MicrocodeDeflattener(ida_hexrays.minsn_visitor_t):
    def __init__(self):
        ida_hexrays.minsn_visitor_t.__init__(self)
        self.state_var = None
        self.state_assignments = {}  # {block_addr: next_state}
        self.dispatcher_addr = None

    def visit_minsn(self):
        """访问每条微码指令"""
        insn = self.curins

        # 查找状态变量赋值: mov state_var, imm
        if insn.opcode == ida_hexrays.m_mov:
            if insn.d.t == ida_hexrays.mop_r:  # 目标是寄存器
                if insn.l.t == ida_hexrays.mop_n:  # 源是立即数
                    state_value = insn.l.nnn.value
                    block_addr = self.blk.head
                    self.state_assignments[block_addr] = state_value

        return 0

def deflattening_pass(cfunc):
    """去平坦化优化 Pass"""
    mba = cfunc.mba

    # 1. 识别状态变量和分发器
    visitor = MicrocodeDeflattener()

    for i in range(mba.qty):
        blk = mba.get_mblock(i)
        visitor.blk = blk
        blk.for_all_insns(visitor)

    print(f"[*] Found {len(visitor.state_assignments)} state assignments")

    # 2. 构建真实的控制流
    real_cfg = {}
    for block_addr, next_state in visitor.state_assignments.items():
        # 找到 next_state 对应的目标块
        for i in range(mba.qty):
            blk = mba.get_mblock(i)
            # 检查块的入口状态
            if matches_state(blk, next_state):
                real_cfg[block_addr] = blk.start
                break

    # 3. Patch 控制流
    for src_addr, dst_addr in real_cfg.items():
        patch_jump(src_addr, dst_addr)

    return cfunc

def matches_state(blk, state_value):
    """检查基本块是否对应某个状态值"""
    # 分析块的条件判断
    tail = blk.tail
    if tail and tail.opcode == ida_hexrays.m_jcnd:
        # 检查条件是否比较状态变量
        pass
    return False

def patch_jump(src_addr, dst_addr):
    """将条件跳转替换为无条件跳转"""
    # 实现 patch 逻辑
    pass

# 注册优化 Pass
# ida_hexrays.install_microcode_filter(deflattening_pass)

  1
  2
  3
  4
  5
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112

# 不透明谓词检测与消除
import ida_hexrays
import idc

class OpaquPredicateDetector:
    # 已知的不透明谓词模式
    OPAQUE_PATTERNS = [
        # (模式描述, 检测函数, 结果)
        ("x * x >= 0", lambda x: True, True),      # 平方数非负
        ("(x & 1) ^ 1 == 0 || (x & 1) == 0", lambda x: True, True),  # 永真
        ("x * (x + 1) % 2 != 0", lambda x: False, False),  # 连续整数乘积是偶数
        ("(x | x) == x", lambda x: True, True),    # 自身或
        ("(x & x) == x", lambda x: True, True),    # 自身与
    ]

    def __init__(self, cfunc):
        self.cfunc = cfunc
        self.opaque_conditions = []

    def analyze(self):
        """分析所有条件表达式"""
        class ConditionVisitor(ida_hexrays.ctree_visitor_t):
            def __init__(self, detector):
                ida_hexrays.ctree_visitor_t.__init__(self,
                    ida_hexrays.CV_FAST)
                self.detector = detector

            def visit_expr(self, expr):
                # 检查 if 条件
                if expr.op == ida_hexrays.cot_if:
                    cond = expr.x
                    if self.detector.is_opaque(cond):
                        self.detector.opaque_conditions.append(expr.ea)
                return 0

        visitor = ConditionVisitor(self)
        visitor.apply_to(self.cfunc.body, None)

        return self.opaque_conditions

    def is_opaque(self, cond_expr):
        """检查条件是否是不透明谓词"""
        # 模式 1: x * x >= 0
        if self.matches_square_pattern(cond_expr):
            return True

        # 模式 2: x == x
        if self.matches_self_comparison(cond_expr):
            return True

        # 模式 3: 常量折叠可判断
        if self.is_constant_foldable(cond_expr):
            return True

        return False

    def matches_square_pattern(self, expr):
        """检测 x * x >= 0 模式"""
        if expr.op == ida_hexrays.cot_sge:  # >=
            left = expr.x
            right = expr.y

            # 右侧是 0
            if right.op == ida_hexrays.cot_num and right.n._value == 0:
                # 左侧是乘法
                if left.op == ida_hexrays.cot_mul:
                    # 乘法两侧相同
                    if self.exprs_equal(left.x, left.y):
                        return True
        return False

    def matches_self_comparison(self, expr):
        """检测 x == x 模式"""
        if expr.op in [ida_hexrays.cot_eq, ida_hexrays.cot_le,
                       ida_hexrays.cot_ge]:
            if self.exprs_equal(expr.x, expr.y):
                return True
        return False

    def exprs_equal(self, e1, e2):
        """检查两个表达式是否相等"""
        if e1.op != e2.op:
            return False

        if e1.op == ida_hexrays.cot_var:
            return e1.v.idx == e2.v.idx

        # 递归检查子表达式
        return False

    def is_constant_foldable(self, expr):
        """检查表达式是否可以常量折叠"""
        # 尝试符号求值
        try:
            result = self.symbolic_eval(expr)
            return result is not None and isinstance(result, bool)
        except:
            return False

    def remove_opaque_predicates(self):
        """移除检测到的不透明谓词"""
        for addr in self.opaque_conditions:
            print(f"[*] Removing opaque predicate at 0x{addr:x}")
            # 根据谓词的恒定结果，简化控制流
            # 永真 -> 删除 else 分支
            # 永假 -> 删除 if 分支

# 使用示例
# cfunc = ida_hexrays.decompile(idc.here())
# detector = OpaquPredicateDetector(cfunc)
# opaques = detector.analyze()
# detector.remove_opaque_predicates()

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69

# 基于 Z3 的表达式化简
from z3 import *

class ExpressionSimplifier:
    def __init__(self):
        self.solver = Solver()

    def simplify_add(self, expr):
        """化简加法替换模式"""
        # 模式: (a ^ b) + 2 * (a & b) -> a + b
        a, b = BitVecs('a b', 32)

        # 原始表达式
        obfuscated = (a ^ b) + 2 * (a & b)

        # 简化形式
        simplified = a + b

        # 验证等价性
        self.solver.add(obfuscated != simplified)
        if self.solver.check() == unsat:
            print("[*] Equivalence verified: (a^b) + 2*(a&b) == a+b")
            return True
        return False

    def simplify_xor(self, expr):
        """化简 XOR 替换模式"""
        a, b = BitVecs('a b', 32)

        # 模式: (~a & b) | (a & ~b) -> a ^ b
        obfuscated = (~a & b) | (a & ~b)
        simplified = a ^ b

        self.solver.reset()
        self.solver.add(obfuscated != simplified)
        if self.solver.check() == unsat:
            print("[*] Equivalence verified: (~a&b)|(a&~b) == a^b")
            return True
        return False

    def pattern_match_and_simplify(self, mba):
        """在 microcode 中匹配和简化混淆模式"""
        simplified_count = 0

        for i in range(mba.qty):
            blk = mba.get_mblock(i)
            insn = blk.head

            while insn:
                # 检测 XOR + AND + LSL + ADD 序列 (加法替换)
                if self.is_add_obfuscation(insn):
                    self.simplify_to_add(insn)
                    simplified_count += 1

                insn = insn.next

        return simplified_count

# 模式替换规则
SUBSTITUTION_RULES = [
    # (混淆模式, 简化结果)
    ("(a ^ b) + 2 * (a & b)", "a + b"),
    ("(a ^ b) + ((a & b) << 1)", "a + b"),
    ("a - (~b + 1)", "a + b"),
    ("(~a & b) | (a & ~b)", "a ^ b"),
    ("(a | b) & (~a | ~b)", "a ^ b"),
    ("~(~a | ~b)", "a & b"),
    ("~(~a & ~b)", "a | b"),
]