Module refinery.lib.emulator.ic

Implements refinery.lib.emulator.interface.RawMetalEmulator for the icicle backend.

Expand source code Browse git 
"""
Implements `refinery.lib.emulator.interface.RawMetalEmulator` for the icicle backend.
"""
from __future__ import annotations

from functools import partial
from typing import TYPE_CHECKING, Any, Callable, TypeVar

from refinery.lib.emulator.abstract import (
    EmulationError,
    EmulationTimeout,
    InvalidInstruction,
    MemAccess,
    RawMetalEmulator,
    Register,
)
from refinery.lib.executable import Arch
from refinery.lib.shared.capstone import capstone
from refinery.lib.shared.icicle import icicle as ic

if TYPE_CHECKING:
    from icicle import Icicle as Ic
else:
    class Ic:
        pass


_T = TypeVar('_T')


class IcicleEmulator(RawMetalEmulator[Ic, str, _T]):
    """
    An Icicle-based emulator. Icicle is a more recent emulator engine and not yet as mature as
    Unicorn. There are some compelling arguments for its robustness, but with the current
    interface it is completely lacking any memory write hook support, which makes it difficult
    to use for most of our applications. See also the [Icicle paper][ICE].

    [ICE]: https://arxiv.org/pdf/2301.13346
    """

    icicle: Ic

    def _init(self):
        super()._init()
        self._single_step = False
        self._halts: set[int] = set()

    def _reset(self):
        super()._reset()
        self._halts.clear()
        exe = self.exe

        try:
            arch = {
                Arch.X32: 'i686',
                Arch.X64: 'x86_64',
            }[exe.arch()]
        except KeyError:
            arch = None
        if arch not in ic.architectures():
            raise NotImplementedError(F'Icicle cannot handle executables of arch {exe.arch().name}')

        self.icicle = ice = ic.Icicle(arch)
        self.regmap = {reg.casefold(): val[1] for reg, val in ice.reg_list().items()}

        self._map_segments()
        self._map_stack_and_heap()

        if self.hooks.ApiCall:
            self._install_api_trampoline()

    def _call_hook(self, hook: Callable[..., Any], *args) -> Any:
        """
        Calls a hook of the emulator. A hook that fails ends the run like a hook of the other
        backends: `refinery.lib.emulator.Emulator.emulate` raises its error.
        """
        try:
            return hook(*args)
        except BaseException as error:
            self._hook_failed(error)
            raise

    def _enable_single_step(self):
        self._single_step = True

    def _disable_single_step(self):
        self._single_step = False

    def _fault_access_size(self, insn) -> int:
        """
        The width of a faulting memory access is the size of the instruction's memory operand;
        an x86 instruction has at most one. When none is present, fall back to the address size.
        """
        for op in insn.operands:
            if op.type == capstone.CS_OP_MEM:
                return op.size
        return insn.addr_size

    def _emulate(self, start: int, end: int | None = None, timeout: int | None = None):
        try:
            return self._run(start, end, timeout)
        finally:
            ice = self.icicle
            for address in self._halts:
                ice.remove_breakpoint(address)
            self._halts.clear()

    def _run(self, start: int, end: int | None, timeout: int | None):
        RS = ic.RunStatus
        MP = ic.MemoryProtection
        ice = self.icicle

        icount_max = (1 << 64) - 1
        start_icount = ice.icount
        if timeout is None:
            ice.icount_limit = icount_max
            deadline = None
        else:
            deadline = start_icount + timeout
            ice.icount_limit = min(deadline, icount_max)

        code_hooked = self.hooks.CodeExecute
        apis_hooked = self.hooks.ApiCall
        mm_e_hooked = self.hooks.MemoryError
        mm_w_hooked = self.hooks.MemoryWrite
        mm_r_hooked = self.hooks.MemoryRead
        mm_x_hooked = mm_r_hooked or mm_w_hooked

        halt = self._single_step
        dasm = self.exe.disassembler()
        dasm.detail = True

        if code_hooked or halt:
            step = partial(ice.step, 1)
        elif end is not None:
            step = partial(ice.run_until, end)
        else:
            step = ice.run

        self.ip = ip = start
        mprotect: list[tuple[int, int, MP]] = []
        retrying = 0
        last_fault: tuple[object, int] | None = None

        while True:
            if end is not None and ip == end:
                break
            if deadline is not None and ice.icount >= deadline:
                raise EmulationTimeout(ice.icount - start_icount)
            if (code_hooked or apis_hooked) and not retrying:
                try:
                    insn = next(dasm.disasm(self.mem_read(ip, 20), ip, 1))
                except StopIteration as SI:
                    raise InvalidInstruction(ip) from SI
                args = (ice, ip, insn.size, self.state)
                if apis_hooked:
                    self._call_hook(self._hook_api_call_check, *args)
                if code_hooked:
                    self._call_hook(self.hook_code_execute, *args)
            else:
                insn = None
            if mprotect:
                ice.mem_protect(*mprotect[-1])
                status = ice.step(1)
            else:
                status = step()
            if status == RS.InstructionLimit:
                stop = False
                for addr, size, prot in mprotect:
                    ice.mem_protect(addr, size, MP.ExecuteOnly)
                    if mm_w_hooked and prot == MP.ExecuteReadWrite:
                        value = self.mem_read_int(addr, size)
                        if self._call_hook(self.hook_mem_write, ice, MemAccess.Write, addr, size, value, self.state) is False:
                            stop = True
                mprotect.clear()
                retrying = 0
                last_fault = None
                ip = self.ip
                if stop:
                    break
            elif status in (
                RS.Breakpoint,
                RS.Halt,
                RS.Killed,
            ):
                break
            elif status == RS.UnhandledException:
                ip = self.ip
                if insn is None or insn.address != ip:
                    insn = next(dasm.disasm(self.mem_read(ip, 20), ip, 1))
                EC = ic.ExceptionCode
                ea = ice.exception_value
                ec = ice.exception_code
                if (ec, ea) == last_fault:
                    raise EmulationError(F'no forward progress at {ea:#x} handling {ec!r}')
                last_fault = (ec, ea)
                size = self._fault_access_size(insn)
                xs = None
                if ec == EC.ReadUnmapped:
                    xs = MemAccess.Unmapped | MemAccess.Read
                if ec == EC.WriteUnmapped:
                    xs = MemAccess.Unmapped | MemAccess.Write
                if xs is not None and mm_e_hooked:
                    if self._call_hook(self.hook_mem_error, ice, xs, ea, size, 0, self.state) is not False:
                        retrying += 1
                        continue
                elif ec == EC.ReadPerm and mm_x_hooked:
                    value = self.mem_read_int(ea, size)
                    if self._call_hook(self.hook_mem_read, ice, MemAccess.Read, ea, size, value, self.state) is not False:
                        prot = MP.ExecuteRead if mm_w_hooked else MP.ExecuteReadWrite
                        mprotect.append((ea, size, prot))
                        retrying += 1
                        continue
                elif ec == EC.WritePerm and mm_x_hooked:
                    mprotect.append((ea, size, MP.ExecuteReadWrite))
                    retrying += 1
                    continue
                else:
                    raise EmulationError(repr(ec))
            elif status != RS.Running:
                raise EmulationError(status.name)
            if halt:
                break

    def halt(self):
        """
        Stops the run at the current instruction with a breakpoint, which the run removes when it
        ends, so that the next run passes it.
        """
        self._halts.add(ip := self.ip)
        self.icicle.add_breakpoint(ip)

    def _lookup_register(self, var: str) -> Register[str]:
        name = var.casefold()
        size = self.regmap[name]
        return Register(name, name, size)

    def _map(self, address: int, size: int):
        MP = ic.MemoryProtection
        if self.hooks.MemoryAccess:
            perm = MP.ExecuteOnly
        else:
            perm = MP.ExecuteReadWrite
        self.icicle.mem_map(address, size, perm)
        return address

    def _set_register(self, register: str, v: int) -> None:
        return self.icicle.reg_write(register, v)

    def _get_register(self, register: str) -> int:
        return self.icicle.reg_read(register)

    def _mem_write(self, address: int, data: bytes):
        return self.icicle.mem_write(address, data)

    def _mem_read(self, address: int, size: int):
        return self.icicle.mem_read(address, size)

Classes

class Ic
Expand source code Browse git 
class Ic:
    pass
class IcicleEmulator (data, base=None, arch=None, hooks=18, align_size=4096, alloc_size=4096, page_limit=268435456)

An Icicle-based emulator. Icicle is a more recent emulator engine and not yet as mature as Unicorn. There are some compelling arguments for its robustness, but with the current interface it is completely lacking any memory write hook support, which makes it difficult to use for most of our applications. See also the Icicle paper.

Expand source code Browse git 
class IcicleEmulator(RawMetalEmulator[Ic, str, _T]):
    """
    An Icicle-based emulator. Icicle is a more recent emulator engine and not yet as mature as
    Unicorn. There are some compelling arguments for its robustness, but with the current
    interface it is completely lacking any memory write hook support, which makes it difficult
    to use for most of our applications. See also the [Icicle paper][ICE].

    [ICE]: https://arxiv.org/pdf/2301.13346
    """

    icicle: Ic

    def _init(self):
        super()._init()
        self._single_step = False
        self._halts: set[int] = set()

    def _reset(self):
        super()._reset()
        self._halts.clear()
        exe = self.exe

        try:
            arch = {
                Arch.X32: 'i686',
                Arch.X64: 'x86_64',
            }[exe.arch()]
        except KeyError:
            arch = None
        if arch not in ic.architectures():
            raise NotImplementedError(F'Icicle cannot handle executables of arch {exe.arch().name}')

        self.icicle = ice = ic.Icicle(arch)
        self.regmap = {reg.casefold(): val[1] for reg, val in ice.reg_list().items()}

        self._map_segments()
        self._map_stack_and_heap()

        if self.hooks.ApiCall:
            self._install_api_trampoline()

    def _call_hook(self, hook: Callable[..., Any], *args) -> Any:
        """
        Calls a hook of the emulator. A hook that fails ends the run like a hook of the other
        backends: `refinery.lib.emulator.Emulator.emulate` raises its error.
        """
        try:
            return hook(*args)
        except BaseException as error:
            self._hook_failed(error)
            raise

    def _enable_single_step(self):
        self._single_step = True

    def _disable_single_step(self):
        self._single_step = False

    def _fault_access_size(self, insn) -> int:
        """
        The width of a faulting memory access is the size of the instruction's memory operand;
        an x86 instruction has at most one. When none is present, fall back to the address size.
        """
        for op in insn.operands:
            if op.type == capstone.CS_OP_MEM:
                return op.size
        return insn.addr_size

    def _emulate(self, start: int, end: int | None = None, timeout: int | None = None):
        try:
            return self._run(start, end, timeout)
        finally:
            ice = self.icicle
            for address in self._halts:
                ice.remove_breakpoint(address)
            self._halts.clear()

    def _run(self, start: int, end: int | None, timeout: int | None):
        RS = ic.RunStatus
        MP = ic.MemoryProtection
        ice = self.icicle

        icount_max = (1 << 64) - 1
        start_icount = ice.icount
        if timeout is None:
            ice.icount_limit = icount_max
            deadline = None
        else:
            deadline = start_icount + timeout
            ice.icount_limit = min(deadline, icount_max)

        code_hooked = self.hooks.CodeExecute
        apis_hooked = self.hooks.ApiCall
        mm_e_hooked = self.hooks.MemoryError
        mm_w_hooked = self.hooks.MemoryWrite
        mm_r_hooked = self.hooks.MemoryRead
        mm_x_hooked = mm_r_hooked or mm_w_hooked

        halt = self._single_step
        dasm = self.exe.disassembler()
        dasm.detail = True

        if code_hooked or halt:
            step = partial(ice.step, 1)
        elif end is not None:
            step = partial(ice.run_until, end)
        else:
            step = ice.run

        self.ip = ip = start
        mprotect: list[tuple[int, int, MP]] = []
        retrying = 0
        last_fault: tuple[object, int] | None = None

        while True:
            if end is not None and ip == end:
                break
            if deadline is not None and ice.icount >= deadline:
                raise EmulationTimeout(ice.icount - start_icount)
            if (code_hooked or apis_hooked) and not retrying:
                try:
                    insn = next(dasm.disasm(self.mem_read(ip, 20), ip, 1))
                except StopIteration as SI:
                    raise InvalidInstruction(ip) from SI
                args = (ice, ip, insn.size, self.state)
                if apis_hooked:
                    self._call_hook(self._hook_api_call_check, *args)
                if code_hooked:
                    self._call_hook(self.hook_code_execute, *args)
            else:
                insn = None
            if mprotect:
                ice.mem_protect(*mprotect[-1])
                status = ice.step(1)
            else:
                status = step()
            if status == RS.InstructionLimit:
                stop = False
                for addr, size, prot in mprotect:
                    ice.mem_protect(addr, size, MP.ExecuteOnly)
                    if mm_w_hooked and prot == MP.ExecuteReadWrite:
                        value = self.mem_read_int(addr, size)
                        if self._call_hook(self.hook_mem_write, ice, MemAccess.Write, addr, size, value, self.state) is False:
                            stop = True
                mprotect.clear()
                retrying = 0
                last_fault = None
                ip = self.ip
                if stop:
                    break
            elif status in (
                RS.Breakpoint,
                RS.Halt,
                RS.Killed,
            ):
                break
            elif status == RS.UnhandledException:
                ip = self.ip
                if insn is None or insn.address != ip:
                    insn = next(dasm.disasm(self.mem_read(ip, 20), ip, 1))
                EC = ic.ExceptionCode
                ea = ice.exception_value
                ec = ice.exception_code
                if (ec, ea) == last_fault:
                    raise EmulationError(F'no forward progress at {ea:#x} handling {ec!r}')
                last_fault = (ec, ea)
                size = self._fault_access_size(insn)
                xs = None
                if ec == EC.ReadUnmapped:
                    xs = MemAccess.Unmapped | MemAccess.Read
                if ec == EC.WriteUnmapped:
                    xs = MemAccess.Unmapped | MemAccess.Write
                if xs is not None and mm_e_hooked:
                    if self._call_hook(self.hook_mem_error, ice, xs, ea, size, 0, self.state) is not False:
                        retrying += 1
                        continue
                elif ec == EC.ReadPerm and mm_x_hooked:
                    value = self.mem_read_int(ea, size)
                    if self._call_hook(self.hook_mem_read, ice, MemAccess.Read, ea, size, value, self.state) is not False:
                        prot = MP.ExecuteRead if mm_w_hooked else MP.ExecuteReadWrite
                        mprotect.append((ea, size, prot))
                        retrying += 1
                        continue
                elif ec == EC.WritePerm and mm_x_hooked:
                    mprotect.append((ea, size, MP.ExecuteReadWrite))
                    retrying += 1
                    continue
                else:
                    raise EmulationError(repr(ec))
            elif status != RS.Running:
                raise EmulationError(status.name)
            if halt:
                break

    def halt(self):
        """
        Stops the run at the current instruction with a breakpoint, which the run removes when it
        ends, so that the next run passes it.
        """
        self._halts.add(ip := self.ip)
        self.icicle.add_breakpoint(ip)

    def _lookup_register(self, var: str) -> Register[str]:
        name = var.casefold()
        size = self.regmap[name]
        return Register(name, name, size)

    def _map(self, address: int, size: int):
        MP = ic.MemoryProtection
        if self.hooks.MemoryAccess:
            perm = MP.ExecuteOnly
        else:
            perm = MP.ExecuteReadWrite
        self.icicle.mem_map(address, size, perm)
        return address

    def _set_register(self, register: str, v: int) -> None:
        return self.icicle.reg_write(register, v)

    def _get_register(self, register: str) -> int:
        return self.icicle.reg_read(register)

    def _mem_write(self, address: int, data: bytes):
        return self.icicle.mem_write(address, data)

    def _mem_read(self, address: int, size: int):
        return self.icicle.mem_read(address, size)

Ancestors

Class variables

var icicle

The type of the None singleton.

Methods

def halt(self)

Stops the run at the current instruction with a breakpoint, which the run removes when it ends, so that the next run passes it.

Expand source code Browse git 
def halt(self):
    """
    Stops the run at the current instruction with a breakpoint, which the run removes when it
    ends, so that the next run passes it.
    """
    self._halts.add(ip := self.ip)
    self.icicle.add_breakpoint(ip)

Inherited members