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, 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

    def _reset(self):
        super()._reset()
        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 _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):
        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._hook_api_call_check(*args)
                if code_hooked:
                    self.hook_code_execute(*args)
            else:
                insn = None
            if mprotect:
                ice.mem_protect(*mprotect[-1])
            if (status := step()) == 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.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:
                insn = insn or 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.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.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):
        self.icicle.add_breakpoint(self.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
        return self.icicle.mem_map(address, size, perm)

    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

    def _reset(self):
        super()._reset()
        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 _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):
        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._hook_api_call_check(*args)
                if code_hooked:
                    self.hook_code_execute(*args)
            else:
                insn = None
            if mprotect:
                ice.mem_protect(*mprotect[-1])
            if (status := step()) == 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.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:
                insn = insn or 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.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.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):
        self.icicle.add_breakpoint(self.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
        return self.icicle.mem_map(address, size, perm)

    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.

Inherited members