#!/usr/bin/env python3 """离线执行 ``libw.so`` 的 ``W.pr(99999, 2, ...)`` 作为差分 oracle。 该工具只服务于逆向和测试向量生成。生产代码不得依赖 Unicorn、ELF 样本或 本模块。运行示例： uv run --with pyelftools --with unicorn \ python -m tools.libw_pr_oracle --payload PAYLOAD """ from __future__ import annotations import argparse import struct from pathlib import Path from typing import Callable DEFAULT_LIBRARY = Path( "out/agent_apk_triage/apktool/lib/arm64-v8a/libw.so" ) PR_ENTRY = 0xECB0 STACK_BASE = 0x70000000 STACK_SIZE = 0x400000 AUX_BASE = 0x71000000 AUX_SIZE = 0x20000 HEAP_BASE = 0x72000000 HEAP_SIZE = 0x1000000 class OracleError(RuntimeError): """离线 native oracle 执行失败。""" def _java_modified_utf8(value: str) -> bytes: utf16 = value.encode("utf-16-be", errors="surrogatepass") output = bytearray() for offset in range(0, len(utf16), 2): code_unit = int.from_bytes(utf16[offset : offset + 2], "big") if 0x01 <= code_unit <= 0x7F: output.append(code_unit) elif code_unit <= 0x7FF: output.extend((0xC0 | (code_unit >> 6), 0x80 | (code_unit & 0x3F))) else: output.extend( ( 0xE0 | (code_unit >> 12), 0x80 | ((code_unit >> 6) & 0x3F), 0x80 | (code_unit & 0x3F), ) ) return bytes(output) def _java_utf16_length(value: str) -> int: return len(value.encode("utf-16-be", errors="surrogatepass")) // 2 class LibwPrOracle: def __init__(self, library: Path = DEFAULT_LIBRARY) -> None: self.library = Path(library) def generate( self, payload: str, code_observer: Callable[[object, int], None] | None = None, format_observer: Callable[[str, int], None] | None = None, ) -> str: """返回与 ``W.pr(99999, 2, len * 2, payload)`` 相同的 KAS。""" try: from elftools.elf.elffile import ELFFile from unicorn import UC_ARCH_ARM64, UC_HOOK_CODE, UC_MODE_LITTLE_ENDIAN, Uc from unicorn.arm64_const import ( UC_ARM64_REG_PC, UC_ARM64_REG_SP, UC_ARM64_REG_TPIDR_EL0, UC_ARM64_REG_X0, UC_ARM64_REG_X1, UC_ARM64_REG_X2, UC_ARM64_REG_X3, UC_ARM64_REG_X4, UC_ARM64_REG_X5, UC_ARM64_REG_X30, ) except ImportError as exc: raise OracleError( "需要通过 `uv run --with pyelftools --with unicorn` 运行 oracle" ) from exc if not self.library.is_file(): raise OracleError(f"libw 样本不存在: {self.library}") with self.library.open("rb") as stream: elf = ELFFile(stream) loads = [ segment for segment in elf.iter_segments() if segment["p_type"] == "PT_LOAD" ] if not loads: raise OracleError("ELF 不含 PT_LOAD 段") page = 0x1000 image_end = max( (int(segment["p_vaddr"]) + int(segment["p_memsz"]) + page - 1) & -page for segment in loads ) emulator = Uc(UC_ARCH_ARM64, UC_MODE_LITTLE_ENDIAN) emulator.mem_map(0, image_end) for segment in loads: emulator.mem_write(int(segment["p_vaddr"]), segment.data()) for section in elf.iter_sections(): if not section.name.startswith(".rela"): continue for relocation in section.iter_relocations(): if relocation.is_RELA(): emulator.mem_write( int(relocation["r_offset"]), struct.pack( " 0x8000: raise OracleError("oracle payload 超过 32767 字节") emulator.mem_write(input_address, payload_bytes + b"\0") emulator.mem_write(env_address, struct.pack(" bytes: if address == 0: return b"" data = bytes(emulator.mem_read(address, limit)) return data.split(b"\0", 1)[0] def return_from_call(value: int = 0) -> None: emulator.reg_write(UC_ARM64_REG_X0, value) emulator.reg_write( UC_ARM64_REG_PC, emulator.reg_read(UC_ARM64_REG_X30), ) def read_va_int(va_list: int) -> int: stack_pointer, gr_top, _vr_top, gr_offs, _vr_offs = struct.unpack( " None: nonlocal heap_next, completed if code_observer is not None: code_observer(emulator, address) if jni_stubs <= address < jni_stubs + 240 * 4: index = (address - jni_stubs) // 4 if index == 169: # GetStringUTFChars return_from_call(input_address) return raise OracleError(f"未实现的 JNI 调用槽位: {index}") name = plt_names.get(address) if name is None: return x0 = emulator.reg_read(UC_ARM64_REG_X0) x1 = emulator.reg_read(UC_ARM64_REG_X1) x2 = emulator.reg_read(UC_ARM64_REG_X2) x3 = emulator.reg_read(UC_ARM64_REG_X3) x4 = emulator.reg_read(UC_ARM64_REG_X4) if name in {"malloc", "calloc"}: size = x0 * x1 if name == "calloc" else x0 result = (heap_next + 15) & -16 heap_next = result + max(size, 16) if heap_next > HEAP_BASE + HEAP_SIZE: raise OracleError("oracle heap 耗尽") emulator.mem_write(result, b"\0" * max(size, 1)) return_from_call(result) return if name == "free": return_from_call() return if name == "memcpy": emulator.mem_write(x0, bytes(emulator.mem_read(x1, x2))) return_from_call(x0) return if name == "memset": emulator.mem_write(x0, bytes([x1 & 0xFF]) * x2) return_from_call(x0) return if name == "strlen": return_from_call(len(read_c_string(x0))) return if name == "__vsprintf_chk": format_string = read_c_string(x3).decode("ascii") value = read_va_int(x4) if format_observer is not None: format_observer(format_string, value) if format_string == "%08x ": rendered = f"{value & 0xFFFFFFFF:08x} ".encode("ascii") elif format_string == "%02x": rendered = f"{value & 0xFF:02x}".encode("ascii") digest.append(value & 0xFF) else: raise OracleError( f"未实现的 __vsprintf_chk 格式: {format_string!r}" ) emulator.mem_write(x0, rendered + b"\0") return_from_call(len(rendered)) if len(digest) == 16: completed = True emulator.emu_stop() return if name == "__stack_chk_fail": raise OracleError("native 栈保护校验失败") raise OracleError(f"未实现的 PLT 调用: {name}") emulator.hook_add(UC_HOOK_CODE, hook_code) try: emulator.emu_start(PR_ENTRY, sentinel, count=10_000_000) except Exception as exc: if isinstance(exc, OracleError): raise raise OracleError( f"native 执行失败，PC={emulator.reg_read(UC_ARM64_REG_PC):#x}: {exc}" ) from exc if not completed or len(digest) != 16: raise OracleError(f"native 未产生完整摘要: {digest.hex()}") return "00" + digest.hex() def main() -> int: parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("--library", type=Path, default=DEFAULT_LIBRARY) parser.add_argument("--payload", required=True) args = parser.parse_args() print(LibwPrOracle(args.library).generate(args.payload)) return 0 if __name__ == "__main__": raise SystemExit(main())