1
2
3
4
5

// 原始操作: X0 = X1 + X2
// MBA 混淆后 (使用 (a | b) + (a & b) = a + b):
AND X3, X1, X2      // X3 = X1 & X2
ORR X4, X1, X2      // X4 = X1 | X2
ADD X0, X3, X4      // X0 = (X1 & X2) + (X1 | X2) = X1 + X2

 1
 2
 3
 4
 5
 6
 7
 8
 9
10

// 原始操作: X0 = X1 + X2
// 深层嵌套 MBA 混淆 (使用 (a ^ b) + 2 * (a & b) = a + b):
EOR X3, X1, X2      // X3 = X1 ^ X2
AND X4, X1, X2      // X4 = X1 & X2
LSL X5, X4, #1      // X5 = (X1 & X2) << 1 = 2 * (X1 & X2)
// 进一步将 X3 + X5 再次 MBA 展开:
EOR X6, X3, X5      // X6 = X3 ^ X5
AND X7, X3, X5      // X7 = X3 & X5
LSL X8, X7, #1      // X8 = 2 * (X3 & X5)
ADD X0, X6, X8      // X0 = X6 + X8 = X3 + X5 = X1 + X2

 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

from sympy import symbols, simplify, And, Or, Xor
from sympy.core.numbers import Integer

def simplify_mba(expr_str):
    """
    简化 MBA 表达式到规范形式
    
    思路：将 MBA 表达式转化为 sympy 符号表达式，
    利用位运算的代数恒等式进行化简。
    """
    a, b = symbols('a b', integer=True)
    
    # 常见 MBA 恒等式验证
    identities = {
        "(a & b) + (a | b)":       a + b,
        "(a ^ b) + 2 * (a & b)":   a + b,
        "(a | b) - (a & b)":       a ^ b,  # XOR
        "(a ^ b) - 2 * (~a & b)":  a - b,
    }
    
    for pattern, simplified in identities.items():
        print(f"  {pattern:40s} => {simplified}")
    
    return identities.get(expr_str, None)

def verify_mba_equivalence():
    """使用穷举法验证 MBA 等价关系（适用于小位宽）"""
    import random
    
    test_cases = [(random.randint(0, 0xFFFFFFFF), random.randint(0, 0xFFFFFFFF)) 
                  for _ in range(10000)]
    
    for a, b in test_cases:
        mask = 0xFFFFFFFF  # 32-bit
        # 验证: (a & b) + (a | b) == a + b
        lhs = ((a & b) + (a | b)) & mask
        rhs = (a + b) & mask
        assert lhs == rhs, f"Failed: a={a:#x}, b={b:#x}"
    
    print("[+] MBA 等价验证通过 (10000 组随机数据)")

verify_mba_equivalence()

1
2
3
4
5
6

.init_array:00001000  DCD decrypt_strings_0  ; 字符串解密函数 1
.init_array:00001004  DCD decrypt_strings_1  ; 字符串解密函数 2
.init_array:00001008  DCD real_init          ; 真正的初始化函数

.rodata:00002000  encrypted_str1  DCB 0x7A, 0x4F, 0x52, ...  ; 加密后的字符串
.rodata:00002020  encrypted_str2  DCB 0x1B, 0x33, 0x6E, ...  ; 加密后的字符串

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

; decrypt_strings_0 - 典型的 XOR 解密循环
decrypt_strings_0:
    ADRP    X0, #encrypted_str1@PAGE
    ADD     X0, X0, #encrypted_str1@PAGEOFF
    MOV     W1, #0x1A          ; 字符串长度
    MOV     W2, #0x5A          ; XOR 密钥
.loop:
    LDRB    W3, [X0]           ; 读取一个加密字节
    EOR     W3, W3, W2         ; XOR 解密
    STRB    W3, [X0], #1       ; 写回并递增指针
    SUBS    W1, W1, #1         ; 计数器减一
    B.NE    .loop              ; 循环直到所有字节解密
    RET

 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

// 方法一: Hook .init_array 函数，在解密完成后 dump 字符串
function dumpDecryptedStrings(soName) {
    var mod = Process.findModuleByName(soName);
    if (!mod) {
        console.log("[-] Module not found: " + soName);
        return;
    }

    // 在 SO 加载完成后，.init_array 已经执行，字符串已解密
    // 直接扫描 .rodata 段寻找可读字符串
    var sections = Process.findModuleByName(soName).enumerateSections();
    sections.forEach(function(section) {
        if (section.name === ".rodata") {
            console.log("[*] Scanning .rodata at: " + section.address + 
                        " size: " + section.size);
            
            // 扫描 ASCII 字符串
            var base = section.address;
            var size = section.size;
            var data = Memory.readByteArray(base, size);
            var bytes = new Uint8Array(data);
            
            var currentStr = "";
            for (var i = 0; i < bytes.length; i++) {
                if (bytes[i] >= 0x20 && bytes[i] < 0x7F) {
                    currentStr += String.fromCharCode(bytes[i]);
                } else {
                    if (currentStr.length >= 4) {
                        console.log("[+] String @ offset 0x" + i.toString(16) + 
                                    ": " + currentStr);
                    }
                    currentStr = "";
                }
            }
        }
    });
}

// 方法二: Hook dlopen，在目标 SO 加载时自动提取
Interceptor.attach(Module.findExportByName(null, "dlopen"), {
    onEnter: function(args) {
        this.path = args[0].readCString();
    },
    onLeave: function(retval) {
        if (this.path && this.path.indexOf("libtarget.so") !== -1) {
            console.log("[*] Target SO loaded, dumping strings...");
            // dlopen 返回时 .init_array 已执行完毕
            setTimeout(function() {
                dumpDecryptedStrings("libtarget.so");
            }, 100);
        }
    }
});

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

// 原始代码 - 直接调用:
BL      target_func          // 直接调用 target_func

// 混淆后 - 间接调用:
ADRP    X8, #func_table@PAGE
ADD     X8, X8, #func_table@PAGEOFF
LDR     X8, [X8, #0x18]     // 从函数表加载目标地址
BLR     X8                   // 间接调用

// 更复杂的混淆 - 带解密的间接调用:
ADRP    X8, #encrypted_ptr@PAGE
LDR     X8, [X8, #encrypted_ptr@PAGEOFF]
MOV     X9, #0xDEAD          // 解密密钥的一部分
MOVK    X9, #0xBEEF, LSL #16
EOR     X8, X8, X9           // XOR 解密得到真实地址
BLR     X8                   // 调用解密后的地址

 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

// Hook 间接调用指令，记录实际跳转目标
function resolveIndirectCalls(soName, funcOffset) {
    var mod = Process.findModuleByName(soName);
    var funcAddr = mod.base.add(funcOffset);
    
    Stalker.follow(Process.getCurrentThreadId(), {
        transform: function(iterator) {
            var inst = iterator.next();
            do {
                // 检测 BLR 指令
                if (inst.mnemonic === "blr") {
                    iterator.putCallout(function(ctx) {
                        // 读取寄存器中的实际目标地址
                        var target = ctx[inst.operands[0].value];
                        var targetMod = Process.findModuleByAddress(target);
                        var symbol = DebugSymbol.fromAddress(target);
                        console.log("[BLR] " + inst.address + " => " + 
                                    target + " (" + symbol.name + ")");
                    });
                }
                iterator.keep();
            } while ((inst = iterator.next()) !== null);
        }
    });
}

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

// Hikari 嵌套分发器的反编译特征：
while (1) {
    switch (state_var_outer) {
        case 0x1A3B:
            switch (state_var_inner) {   // ← 嵌套 switch，这是 Hikari 的标志
                case 0x7F2C:
                    // 真实代码块
                    state_var_inner = 0x3D1E;
                    break;
                case 0x3D1E:
                    // 另一个真实代码块
                    break;
            }
            state_var_outer = 0x5C4D;
            break;
        case 0x5C4D:
            // ...
            break;
    }
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10

// Pluto-Obfuscator 的 IndirectBranch 特征：
// 大量使用 BLR/BR 配合从内存加载的地址
ADRP    X16, #jump_table@PAGE
ADD     X16, X16, #jump_table@PAGEOFF
LDRSW   X17, [X16, X8, LSL #2]   // 从跳转表加载偏移
ADD     X16, X16, X17             // 计算目标地址
BR      X16                       // 间接跳转

// IDA 反编译中表现为大量的函数指针调用:
((void (*)(void))(base + offset_table[state]))();

 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

# Android 10+ XOM 环境下通过 Frida 读取代码段
# 标准 Memory.readByteArray 可能因 XOM 失败
# 使用以下策略绕过

def read_code_with_frida(base_offset, size):
    """
    通过 Frida 的 ptrace 能力绕过 XOM 读取代码
    XOM 仅阻止进程自身的 load 指令读取，
    但 Frida 通过 ptrace 附加的方式可以绕过此限制
    """
    script_code = """
    'use strict';
    
    var soName = 'libtarget.so';
    var mod = Process.findModuleByName(soName);
    
    if (mod === null) {
        send({error: 'Module not found'});
    } else {
        try {
            // 方法1: 直接读取（在部分设备上 Frida 可以绕过 XOM）
            var code = Memory.readByteArray(mod.base.add(%d), %d);
            send({type: 'code', data: code});
        } catch (e) {
            // 方法2: 如果直接读取失败，使用 mprotect 修改权限
            try {
                Memory.protect(mod.base.add(%d), %d, 'rwx');
                var code = Memory.readByteArray(mod.base.add(%d), %d);
                send({type: 'code', data: code});
            } catch (e2) {
                send({error: 'Cannot read code: ' + e2.message});
            }
        }
    }
    """ % (base_offset, size, base_offset, size, base_offset, size)
    
    return script_code

 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

// 自动检测 XOM 并选择合适的读取策略
function adaptiveCodeRead(soName, offset, size) {
    var mod = Process.findModuleByName(soName);
    if (!mod) { return null; }
    
    var addr = mod.base.add(offset);
    
    // 尝试读取 - 检测是否有 XOM
    try {
        var data = Memory.readByteArray(addr, Math.min(size, 16));
        // 成功，没有 XOM 或 Frida 已绕过
        console.log("[*] Direct read succeeded (no XOM or bypassed)");
        return Memory.readByteArray(addr, size);
    } catch (e) {
        console.log("[!] XOM detected: " + e.message);
        
        // 方案 A: 尝试修改内存保护属性
        try {
            Memory.protect(addr, size, 'rwx');
            console.log("[*] Memory protection changed to rwx");
            return Memory.readByteArray(addr, size);
        } catch (e2) {
            console.log("[!] mprotect failed: " + e2.message);
        }
        
        // 方案 B: 从 /proc/self/mem 读取
        try {
            var fd = new File("/proc/self/mem", "rb");
            fd.seek(addr.toInt32());
            var data = fd.readBytes(size);
            fd.close();
            console.log("[*] Read via /proc/self/mem succeeded");
            return data;
        } catch (e3) {
            console.log("[!] /proc/self/mem read failed: " + e3.message);
        }
        
        // 方案 C: 从磁盘文件读取（不受 XOM 影响）
        console.log("[*] Falling back to disk read");
        return readFromDisk(mod.path, offset, size);
    }
}

function readFromDisk(path, offset, size) {
    var fd = new File(path, "rb");
    fd.seek(offset);
    var data = fd.readBytes(size);
    fd.close();
    return data;
}

1
2
3
4
5
6
7

// 4KB 页面 (传统) - SO 文件段对齐
LOAD  offset=0x000000  vaddr=0x000000  align=0x1000  (4KB)
LOAD  offset=0x001000  vaddr=0x001000  align=0x1000  (4KB)

// 16KB 页面 (Android 15+) - SO 文件段对齐
LOAD  offset=0x000000  vaddr=0x000000  align=0x4000  (16KB)
LOAD  offset=0x004000  vaddr=0x004000  align=0x4000  (16KB)

1
2
3

# 在 ADB shell 中检测页面大小
adb shell getconf PAGE_SIZE
# 输出 4096 (4KB) 或 16384 (16KB)

1

pip install z3-solver

1
2
3
4
5
6
7
8
9

// 混淆后的代码
int x = get_input();
int y = x * x;
if ((y % 2) == 1) {  // 平方数永远不可能是奇数（对于整数）
    // 这个分支永远不会执行（死代码）
    fake_path();
} else {
    real_path();
}

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

from z3 import *

def prove_opaque_predicate():
    """证明 x*x % 2 == 1 对于任意整数 x 都是假的"""
    x = BitVec('x', 32)  # 32位整数
    y = x * x            # y = x^2

    # 创建求解器
    solver = Solver()

    # 尝试找到使 y % 2 == 1 成立的 x
    solver.add(URem(y, 2) == 1)

    result = solver.check()
    if result == unsat:
        print("[+] 证明成功：条件 (x*x % 2 == 1) 恒假")
        print("    这是一个不透明谓词，对应的分支是死代码")
    elif result == sat:
        print("[-] 找到反例:", solver.model())
    else:
        print("[?] 无法确定")

prove_opaque_predicate()

 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

from z3 import *

def analyze_complex_predicate():
    """
    分析复杂的不透明谓词：
    (x * (x + 1)) % 2 == 0  恒真（连续两个整数的乘积必为偶数）
    """
    x = BitVec('x', 32)

    # 表达式：x * (x + 1)
    expr = x * (x + 1)

    solver = Solver()
    # 尝试找到使 expr % 2 != 0 的情况
    solver.add(URem(expr, 2) != 0)

    if solver.check() == unsat:
        print("[+] 证明：(x * (x + 1)) % 2 == 0 恒真")
        print("    这个 if 分支总是会执行")
    else:
        print("[-] 找到反例:", solver.model())

def analyze_bcf_condition():
    """
    分析 OLLVM BCF 生成的典型条件：
    ((x & 0xFFFFFFFE) * (x | 1)) % 2 == 0
    """
    x = BitVec('x', 32)

    # OLLVM BCF 典型模式
    a = x & 0xFFFFFFFE  # 清除最低位，保证是偶数
    b = x | 1           # 设置最低位，保证是奇数
    product = a * b     # 偶数 * 奇数 = 偶数

    solver = Solver()
    solver.add(URem(product, 2) != 0)

    if solver.check() == unsat:
        print("[+] BCF 条件恒真：可以安全移除 else 分支")
    else:
        print("[-] 条件不恒定")

analyze_complex_predicate()
analyze_bcf_condition()

 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

from z3 import *

class OpaquePredicateAnalyzer:
    """批量分析 OLLVM 不透明谓词"""

    # 常见的 OLLVM 不透明谓词模式
    PATTERNS = {
        "square_mod_2": lambda x: URem(x * x, 2) == 1,           # 恒假
        "consecutive_product": lambda x: URem(x * (x + 1), 2) != 0,  # 恒假
        "cubic_identity": lambda x: URem(x * x * x - x, 6) != 0,     # 恒假 (n³-n 能被6整除)
    }

    def __init__(self, bits=32):
        self.bits = bits

    def analyze_all(self):
        """分析所有已知的不透明谓词模式"""
        x = BitVec('x', self.bits)

        print("=" * 60)
        print("OLLVM 不透明谓词分析报告")
        print("=" * 60)

        for name, predicate in self.PATTERNS.items():
            solver = Solver()
            solver.add(predicate(x))

            result = solver.check()
            status = "恒假 (死代码)" if result == unsat else "可能为真"
            print(f"\n[{name}]")
            print(f"  结果: {status}")

            if result == sat:
                print(f"  反例: x = {solver.model()[x]}")

    def analyze_custom(self, condition_func, name="custom"):
        """分析自定义条件"""
        x = BitVec('x', self.bits)
        solver = Solver()
        solver.add(condition_func(x))

        result = solver.check()
        if result == unsat:
            return f"{name}: 恒假"
        elif result == sat:
            return f"{name}: 可满足, 反例 x={solver.model()[x]}"
        else:
            return f"{name}: 未知"

# 使用示例
analyzer = OpaquePredicateAnalyzer()
analyzer.analyze_all()

1

pip install angr

 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

import angr
import claripy

def simple_symbolic_execution(binary_path, target_addr, avoid_addrs=None):
    """
    使用符号执行找到到达目标地址的输入

    Args:
        binary_path: SO 文件路径
        target_addr: 目标地址（如解密函数返回点）
        avoid_addrs: 需要避开的地址列表（如错误处理分支）
    """
    # 加载二进制文件
    proj = angr.Project(binary_path, auto_load_libs=False)

    # 创建符号化的输入（假设输入是一个 32 字节的 buffer）
    sym_input = claripy.BVS('input', 32 * 8)

    # 创建初始状态
    state = proj.factory.entry_state(
        args=[binary_path],
        stdin=angr.SimFile('/dev/stdin', content=sym_input)
    )

    # 创建模拟管理器
    simgr = proj.factory.simulation_manager(state)

    # 探索到目标地址
    simgr.explore(find=target_addr, avoid=avoid_addrs or [])

    if simgr.found:
        found_state = simgr.found[0]
        # 获取满足条件的具体输入
        solution = found_state.solver.eval(sym_input, cast_to=bytes)
        print(f"[+] 找到有效输入: {solution.hex()}")
        return solution
    else:
        print("[-] 未找到有效路径")
        return None

  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

import angr
import claripy

class OLLVMDeobfuscator:
    """OLLVM 反混淆器 - 使用 Angr 符号执行"""

    def __init__(self, binary_path, base_addr=0x0):
        self.proj = angr.Project(
            binary_path,
            main_opts={'base_addr': base_addr},
            auto_load_libs=False
        )
        self.cfg = None

    def build_cfg(self):
        """构建控制流图（用于分析混淆结构）"""
        print("[*] 正在构建 CFG...")
        self.cfg = self.proj.analyses.CFGFast()
        print(f"[+] CFG 构建完成: {len(self.cfg.graph.nodes())} 个节点")
        return self.cfg

    def find_dispatcher(self, func_addr):
        """
        识别控制流平坦化的分发器
        特征：大量的 case 分支，状态变量比较
        """
        func = self.cfg.functions.get(func_addr)
        if not func:
            return None

        # 找到入度最高的基本块（通常是分发器）
        max_in_degree = 0
        dispatcher = None

        for block in func.blocks:
            in_degree = len(list(self.cfg.graph.predecessors(block)))
            if in_degree > max_in_degree:
                max_in_degree = in_degree
                dispatcher = block

        if dispatcher and max_in_degree > 5:
            print(f"[+] 疑似分发器: 0x{dispatcher.addr:x} (入度: {max_in_degree})")
            return dispatcher
        return None

    def trace_execution(self, start_addr, input_data, max_steps=10000):
        """
        符号执行追踪，记录真实执行的基本块
        """
        state = self.proj.factory.blank_state(addr=start_addr)

        # 设置符号化输入
        sym_input = claripy.BVS('input', len(input_data) * 8)
        state.memory.store(state.regs.rdi, sym_input)  # 假设第一个参数是输入

        # 记录执行的基本块
        executed_blocks = []

        def block_hook(state):
            executed_blocks.append(state.addr)

        # 设置 Hook
        self.proj.hook(start_addr, block_hook, length=0)

        simgr = self.proj.factory.simulation_manager(state)
        simgr.run(n=max_steps)

        return executed_blocks

    def symbolic_execution_with_constraints(self, func_addr, target_output):
        """
        通过约束求解，找到产生特定输出的输入
        适用于分析加密/签名算法
        """
        state = self.proj.factory.call_state(func_addr)

        # 创建符号化参数
        sym_arg1 = claripy.BVS('arg1', 64)
        sym_arg2 = claripy.BVS('arg2', 64)

        # 设置函数参数 (x86_64 调用约定)
        state.regs.rdi = sym_arg1
        state.regs.rsi = sym_arg2

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

        # 探索所有路径
        simgr.run()

        # 在结束状态中查找满足输出约束的
        for deadended in simgr.deadended:
            # 假设返回值在 rax
            deadended.solver.add(deadended.regs.rax == target_output)

            if deadended.solver.satisfiable():
                arg1_val = deadended.solver.eval(sym_arg1)
                arg2_val = deadended.solver.eval(sym_arg2)
                print(f"[+] 找到输入: arg1=0x{arg1_val:x}, arg2=0x{arg2_val:x}")
                return (arg1_val, arg2_val)

        return None

# 使用示例
def example_usage():
    # 加载混淆的 SO 文件
    deobf = OLLVMDeobfuscator("libencrypt.so", base_addr=0x10000)

    # 构建 CFG
    deobf.build_cfg()

    # 查找分发器
    dispatcher = deobf.find_dispatcher(0x12340)

    # 符号执行找到产生特定签名的输入
    result = deobf.symbolic_execution_with_constraints(
        func_addr=0x12340,
        target_output=0xDEADBEEF
    )

  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 angr
from angr.analyses.decompiler.condition_processor import ConditionProcessor

class CFGRecovery:
    """恢复被 OLLVM 平坦化的控制流"""

    def __init__(self, proj, func_addr):
        self.proj = proj
        self.func_addr = func_addr
        self.state_var = None
        self.real_blocks = {}  # state_value -> block_addr
        self.transitions = {}  # (from_state, to_state)

    def identify_state_variable(self, dispatcher_addr):
        """
        识别控制流平坦化的状态变量
        状态变量特征：
        1. 在分发器开头被加载
        2. 用于 switch-case 比较
        3. 在每个真实块末尾被更新
        """
        block = self.proj.factory.block(dispatcher_addr)

        # 分析 VEX IR 找到状态变量
        for stmt in block.vex.statements:
            # 查找从内存加载的操作
            if hasattr(stmt, 'data') and hasattr(stmt.data, 'tag'):
                if stmt.data.tag == 'Iex_Load':
                    # 这可能是状态变量
                    print(f"[*] 疑似状态变量加载: {stmt}")

        return self.state_var

    def extract_real_blocks(self, cfg):
        """
        从平坦化的 CFG 中提取真实的基本块
        真实块特征：
        1. 不是分发器
        2. 会修改状态变量
        3. 跳转回分发器
        """
        func = cfg.functions.get(self.func_addr)
        real_blocks = []

        for block in func.blocks:
            # 检查是否跳回分发器（平坦化的特征）
            successors = list(cfg.graph.successors(block))

            # 分析块中的状态变量修改
            # 这里需要更详细的数据流分析

            real_blocks.append(block)

        return real_blocks

    def recover_transitions(self, blocks):
        """
        恢复真实块之间的转换关系
        通过符号执行确定每个块的后继
        """
        transitions = []

        for block in blocks:
            state = self.proj.factory.blank_state(addr=block.addr)

            # 符号执行这个块
            simgr = self.proj.factory.simulation_manager(state)
            simgr.step()

            # 分析状态变量的新值来确定后继
            for succ_state in simgr.active:
                # 读取状态变量的值
                # new_state = succ_state.memory.load(state_var_addr, 4)
                pass

        return transitions

    def rebuild_cfg(self):
        """
        重建原始的控制流图
        """
        # 1. 识别状态变量
        # 2. 提取真实块
        # 3. 恢复转换关系
        # 4. 构建新的 CFG
        pass

def deflat_function(binary_path, func_addr):
    """
    反平坦化函数的完整流程
    """
    proj = angr.Project(binary_path, auto_load_libs=False)

    print(f"[*] 分析函数 @ 0x{func_addr:x}")

    # 构建 CFG
    cfg = proj.analyses.CFGFast()

    # 创建恢复器
    recovery = CFGRecovery(proj, func_addr)

    # 提取真实块
    real_blocks = recovery.extract_real_blocks(cfg)
    print(f"[+] 识别到 {len(real_blocks)} 个真实基本块")

    # 恢复转换
    transitions = recovery.recover_transitions(real_blocks)

    # 重建 CFG
    recovery.rebuild_cfg()

    return recovery

1
2
3
4
5

Step 1: Frida hook 签名函数 → 收集 3+ 组 (输入, 输出) 样本
Step 2: Z3 约束求解 → 假设算法结构，尝试推断密钥
Step 3: 如果 Z3 失败 → Angr 符号执行，追踪完整计算过程
Step 4: 根据追踪结果 → 生成等价 Python 代码
Step 5: 用收集的样本验证 → 确认还原正确性

 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

// collect_samples.js — 收集签名函数的输入输出
// 使用: frida -U -f com.example.app -l collect_samples.js --no-pause

Java.perform(function() {
    // 方法 1: Hook Java 层签名方法
    var SignHelper = Java.use("com.example.app.SignHelper");
    SignHelper.getSign.implementation = function(data) {
        var result = this.getSign(data);
        console.log(JSON.stringify({
            input: data,
            output: result,
            input_hex: stringToHex(data),
        }));
        return result;
    };
    
    // 方法 2: Hook Native 函数（如果已知偏移）
    var mod = Process.findModuleByName("libsign.so");
    if (mod) {
        // 假设函数接受 (char* data, int len) 返回 int
        Interceptor.attach(mod.base.add(0x1234), {
            onEnter: function(args) {
                this.input = args[0].readCString();
                this.len = args[1].toInt32();
            },
            onLeave: function(retval) {
                console.log("[SAMPLE] input=" + this.input + 
                           " len=" + this.len + 
                           " output=0x" + retval.toInt32().toString(16));
            }
        });
    }
});

function stringToHex(str) {
    var hex = '';
    for (var i = 0; i < str.length; i++) {
        hex += str.charCodeAt(i).toString(16).padStart(2, '0');
    }
    return hex;
}

1
2
3
4
5
6
7

# 运行收集脚本，然后在 App 中多次触发签名操作
frida -U -f com.example.app -l collect_samples.js --no-pause

# 预期输出 (收集至少 3 组):
# [SAMPLE] input=test1 len=5 output=0xAABBCCDD
# [SAMPLE] input=test2 len=5 output=0x11223344  
# [SAMPLE] input=a     len=1 output=0xDEADBEEF

  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

import angr
import claripy
from z3 import *

class SignatureAnalyzer:
    """分析 OLLVM 混淆的签名算法"""

    def __init__(self, so_path, sign_func_offset):
        self.proj = angr.Project(so_path, auto_load_libs=False)
        self.sign_func = self.proj.loader.main_object.mapped_base + sign_func_offset

    def analyze_with_known_io(self, known_inputs, known_outputs):
        """
        使用已知的输入输出对来推断算法

        Args:
            known_inputs: 已知输入列表
            known_outputs: 对应的输出列表
        """
        # 使用 Z3 建立约束
        solver = Solver()

        # 假设签名算法是 output = (input * key1 + key2) ^ key3
        key1 = BitVec('key1', 32)
        key2 = BitVec('key2', 32)
        key3 = BitVec('key3', 32)

        for inp, out in zip(known_inputs, known_outputs):
            inp_bv = BitVecVal(inp, 32)
            out_bv = BitVecVal(out, 32)
            solver.add((inp_bv * key1 + key2) ^ key3 == out_bv)

        if solver.check() == sat:
            model = solver.model()
            print(f"[+] 推断出密钥:")
            print(f"    key1 = 0x{model[key1].as_long():08x}")
            print(f"    key2 = 0x{model[key2].as_long():08x}")
            print(f"    key3 = 0x{model[key3].as_long():08x}")
            return model
        else:
            print("[-] 无法推断算法")
            return None

    def symbolic_trace(self, input_value):
        """
        符号执行追踪签名函数
        """
        state = self.proj.factory.call_state(
            self.sign_func,
            input_value,  # 第一个参数
            0,            # 第二个参数（如长度）
        )

        # 记录所有的算术操作
        operations = []

        def track_operations(state):
            # 记录当前执行的指令
            block = state.block()
            for insn in block.capstone.insns:
                if insn.mnemonic in ['xor', 'add', 'sub', 'mul', 'shl', 'shr', 'and', 'or']:
                    operations.append({
                        'addr': insn.address,
                        'op': insn.mnemonic,
                        'operands': insn.op_str
                    })

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

        # 逐步执行并记录
        while simgr.active:
            for s in simgr.active:
                track_operations(s)
            simgr.step()

        return operations

    def generate_equivalent_code(self, operations):
        """
        根据追踪结果生成等价的 Python 代码
        """
        code_lines = ["def sign(input_val):"]
        code_lines.append("    result = input_val")

        for op in operations:
            if op['op'] == 'xor':
                code_lines.append(f"    result ^= ...  # @ 0x{op['addr']:x}")
            elif op['op'] == 'add':
                code_lines.append(f"    result += ...  # @ 0x{op['addr']:x}")
            # ... 其他操作

        code_lines.append("    return result")
        return '\n'.join(code_lines)


# 实战使用示例
def real_world_example():
    """
    实战：分析某 App 的签名算法
    """
    # 1. 首先用 Frida 收集几组输入输出
    known_data = [
        (0x12345678, 0xAABBCCDD),
        (0x87654321, 0x11223344),
        (0x00000001, 0xDEADBEEF),
    ]

    # 2. 使用 Z3 推断可能的算法结构
    analyzer = SignatureAnalyzer("libsign.so", 0x1234)

    inputs = [d[0] for d in known_data]
    outputs = [d[1] for d in known_data]

    # 尝试推断密钥
    keys = analyzer.analyze_with_known_io(inputs, outputs)

    # 3. 如果简单推断失败，使用符号执行
    if not keys:
        print("[*] 切换到符号执行模式...")
        operations = analyzer.symbolic_trace(0x12345678)

        print("[+] 检测到的核心操作:")
        for op in operations:
            print(f"    0x{op['addr']:x}: {op['op']} {op['operands']}")

        # 生成等价代码
        code = analyzer.generate_equivalent_code(operations)
        print("\n[+] 等价 Python 代码:")
        print(code)

if __name__ == "__main__":
    real_world_example()

1
2
3
4
5
6
7
8

# 克隆仓库
git clone https://github.com/joydo/d810

# 将 d810 目录复制到 IDA 插件目录
# macOS:  ~/Library/Application Support/Hex-Rays/IDA Pro/plugins/
# Linux:  ~/.idapro/plugins/
# Windows: %APPDATA%/Hex-Rays/IDA Pro/plugins/
cp -r d810 /path/to/ida/plugins/

 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

# d810 自定义规则: 简化 MBA 模式 (a & b) + (a | b) => a + b
# 文件位置: d810/optimizers/custom_mba.py

from d810.optimizers.flow import MicroCodeOptimizer
from d810.tracker import MicroCodeTracker
import ida_hexrays as hr

class CustomMBASimplifier(MicroCodeOptimizer):
    """
    自定义 MBA 简化规则
    匹配模式: (x & y) + (x | y)
    替换为:   x + y
    """
    
    OPTIMIZER_NAME = "Custom MBA: (x&y)+(x|y) => x+y"
    
    def check_candidate(self, blk, ins):
        """检查当前指令是否匹配 MBA 模式"""
        
        # 匹配 ADD 指令
        if ins.opcode != hr.m_add:
            return False
        
        left = ins.l   # 左操作数
        right = ins.r  # 右操作数
        
        # 检查左操作数是否为 AND 操作
        if not self._is_insn(left, hr.m_and):
            return False
        
        # 检查右操作数是否为 OR 操作
        if not self._is_insn(right, hr.m_or):
            return False
        
        # 获取 AND 和 OR 的操作数
        and_left, and_right = self._get_operands(left)
        or_left, or_right = self._get_operands(right)
        
        # 检查操作数是否相同 (x&y 和 x|y 中的 x,y 相同)
        if self._same_operands(and_left, or_left) and \
           self._same_operands(and_right, or_right):
            self.x = and_left
            self.y = and_right
            return True
        
        return False
    
    def apply_replacement(self, blk, ins):
        """将匹配的 MBA 模式替换为简化形式"""
        # 替换为 x + y
        ins.opcode = hr.m_add
        ins.l = self.x
        ins.r = self.y
        return True

# 更多自定义规则: (x ^ y) + 2*(x & y) => x + y
class CustomMBASimplifier2(MicroCodeOptimizer):
    OPTIMIZER_NAME = "Custom MBA: (x^y)+2*(x&y) => x+y"
    
    def check_candidate(self, blk, ins):
        if ins.opcode != hr.m_add:
            return False
        
        # 匹配 (x ^ y) + (x & y) << 1
        # 或 (x ^ y) + 2 * (x & y)
        # ... (匹配逻辑类似上述规则)
        return False