mirror of
https://github.com/mandiant/speakeasy.git
synced 2026-10-12 02:18:47 +00:00
Willi Ballenthin
4730f34fbe
Synthetic API modules were decoy PE husks whose export stubs only returned an index. Import binding and GetProcAddress gave out sentinel addresses such as 0xFEEDFACE outside any mapping, and the emulator caught calls through the invalid-fetch callback and a code hook on each module. The export table and the IAT therefore gave different addresses for one function, and a guest or debugger could not read, hook, patch or break on the address of an imported API. A module without a handler loaded the default_exe or default_driver decoy PE. One session API registry now owns every synthesized entry. ApiModuleLoader in api_image.py builds a mapped image for each API module, from the exact declarations of the signature database and the explicit function and data handlers. The export table, the IAT and the resolvers give the same mapped entry for a function. Each function entry jumps to a private trap token in a permanently unmapped 1 MiB reservation. A fetch of a token stops Unicorn, and _run_api_engine dispatches the handler outside the native callback. Other accesses to the reservation raise an access violation in the guest. The pc of an API event is the call instruction when memory tracing is enabled, and the caller's return address otherwise. A module without a handler gets an empty placeholder image with a dynamic arena. The get_proc and resolve_export resolvers back GetProcAddress, LdrGetProcedureAddress and MmGetSystemRoutineAddress. They follow forwarders and create dynamic-only entries in the .dyn arena of a synthetic module. Import binding works on each IAT slot: a missing synthetic export gets a dynamic entry, and an invalid slot keeps its bytes. Data exports have writable storage in .data, and the CRT _fmode, _commode and __initenv exports share it with their pointer accessors. The per-run timeout and max_instructions limits now count across API yields, with exact instruction accounting when a limit is set. The sentinel addresses, import_table, impdata_queue, the module access hook and the default_exe/default_driver fallback are removed. Handler callbacks always run inside an API frame, so setup_callback no longer takes caller arguments, and the unused _normalize_mod_name is removed. A guest PE without relocation data cannot be rebased. get_symbols returns a snapshot, GDB breakpoints use the public entries, and doc/unified-api-addresses.md describes the design.
115 lines
4.5 KiB
Python
115 lines
4.5 KiB
Python
"""Nested guest callbacks from API handlers return to the originating API frames."""
|
|
|
|
import struct
|
|
|
|
import pytest
|
|
|
|
from speakeasy import Speakeasy
|
|
|
|
OUTER_PARAM = 0x1111
|
|
LEAF_PARAM = 0x2222
|
|
|
|
|
|
def x86_program(base, data, api):
|
|
"""
|
|
Return code that calls ``api`` (CreateDialogIndirectParamA) with an outer
|
|
dialog procedure, which calls ``api`` again with a leaf procedure. Each
|
|
procedure stores its HWND and lParam in ``data``; the caller stores its
|
|
stack delta and the API result.
|
|
"""
|
|
|
|
def d(value):
|
|
return struct.pack("<I", value)
|
|
|
|
def call_api(proc, param):
|
|
return b"\x68" + d(param) + b"\x68" + d(proc) + b"\x6a\x00" * 3 + b"\xb8" + d(api) + b"\xff\xd0"
|
|
|
|
def record_args(slot):
|
|
code = b"\x8b\x44\x24\x10\xa3" + d(data + slot)
|
|
return code + b"\x8b\x44\x24\x04\xa3" + d(data + slot + 8)
|
|
|
|
outer, leaf = base + 0x100, base + 0x200
|
|
main = b"\x89\xe6" + call_api(outer, OUTER_PARAM) + b"\x29\xe6\x89\x35" + d(data) + b"\xa3" + d(data + 8) + b"\xc3"
|
|
finish = b"\xb8\x01\x00\x00\x00\xc2\x10\x00"
|
|
outer_code = record_args(0x10) + call_api(leaf, LEAF_PARAM) + b"\xa3" + d(data + 0x40) + finish
|
|
leaf_code = record_args(0x20) + finish
|
|
return {base: main, outer: outer_code, leaf: leaf_code}
|
|
|
|
|
|
def x64_program(base, data, api):
|
|
def q(value):
|
|
return struct.pack("<Q", value)
|
|
|
|
def call_api(proc, param):
|
|
code = b"\x48\x83\xec\x38\x48\xc7\x44\x24\x20" + struct.pack("<I", param)
|
|
code += b"\x31\xc9\x31\xd2\x45\x31\xc0\x49\xb9" + q(proc) + b"\x48\xb8" + q(api)
|
|
return code + b"\xff\xd0\x48\x83\xc4\x38"
|
|
|
|
def record_args(slot):
|
|
return b"\x49\xba" + q(data) + b"\x4d\x89\x4a" + bytes([slot]) + b"\x49\x89\x4a" + bytes([slot + 8])
|
|
|
|
outer, leaf = base + 0x100, base + 0x200
|
|
main = b"\x48\x89\xe6" + call_api(outer, OUTER_PARAM) + b"\x48\x29\xe6"
|
|
main += b"\x48\xbb" + q(data) + b"\x48\x89\x33\x48\x89\x43\x08\xc3"
|
|
finish = b"\xb8\x01\x00\x00\x00\xc3"
|
|
outer_code = record_args(0x10) + call_api(leaf, LEAF_PARAM)
|
|
outer_code += b"\x49\xba" + q(data) + b"\x49\x89\x42\x40" + finish
|
|
leaf_code = record_args(0x20) + finish
|
|
return {base: main, outer: outer_code, leaf: leaf_code}
|
|
|
|
|
|
@pytest.mark.parametrize("architecture", ["x86", "amd64"])
|
|
def test_nested_dialog_callbacks_return_to_their_api_callers(config, architecture):
|
|
with Speakeasy(config=config) as se:
|
|
base = se.load_shellcode(data=b"\xcc" * 0x1000, arch=architecture)
|
|
data = se.mem_alloc(0x1000)
|
|
api = se.emu.get_proc("user32", "CreateDialogIndirectParamA")
|
|
program = x86_program if architecture == "x86" else x64_program
|
|
for address, code in program(base, data, api).items():
|
|
se.mem_write(address, code)
|
|
|
|
se.run_shellcode(base)
|
|
|
|
run = se.get_report().entry_points[0]
|
|
width = se.get_ptr_size()
|
|
|
|
def slot(offset):
|
|
return int.from_bytes(se.mem_read(data + offset, width), "little")
|
|
|
|
assert run.error is None
|
|
assert [event.api_name for event in run.events if event.event == "api"] == [
|
|
"user32.CreateDialogIndirectParamA"
|
|
] * 2
|
|
assert slot(0) == 0
|
|
assert (slot(0x10), slot(0x20)) == (OUTER_PARAM, LEAF_PARAM)
|
|
outer_hwnd, leaf_hwnd = slot(0x18), slot(0x28)
|
|
assert outer_hwnd and leaf_hwnd and outer_hwnd != leaf_hwnd
|
|
assert slot(0x40) == leaf_hwnd
|
|
assert slot(8) == run.ret_val == outer_hwnd
|
|
|
|
|
|
@pytest.mark.parametrize("argc", [4, 5, 6])
|
|
def test_x64_api_callback_entry_stack_is_aligned(config, argc):
|
|
with Speakeasy(config=config) as se:
|
|
base = se.load_shellcode(data=b"\xcc" * 0x1000, arch="amd64")
|
|
data = se.mem_alloc(0x1000)
|
|
api = se.emu.get_proc("kernel32", "GetTickCount")
|
|
callback = base + 0x100
|
|
|
|
def handler(_se, _api, _original, args):
|
|
se.emu.api.load_api_handler("user32").setup_callback(callback, list(range(argc)))
|
|
return 0
|
|
|
|
se.add_api_hook(handler, "kernel32", "GetTickCount", argc=0)
|
|
main = b"\x48\x83\xec\x28\x48\xb8" + struct.pack("<Q", api) + b"\xff\xd0\x48\x83\xc4\x28\xc3"
|
|
# mov r10, data; mov [r10], rsp; ret
|
|
record_rsp = b"\x49\xba" + struct.pack("<Q", data) + b"\x49\x89\x22\xc3"
|
|
se.mem_write(base, main)
|
|
se.mem_write(callback, record_rsp)
|
|
|
|
se.run_shellcode(base)
|
|
|
|
assert se.get_report().entry_points[0].error is None
|
|
rsp = int.from_bytes(se.mem_read(data, 8), "little")
|
|
assert rsp and (rsp + 8) % 16 == 0
|