manticore/scripts/verify.py

from __future__ import print_function
from manticore.native import Manticore
from manticore.platforms import linux_syscalls

import logging

from sys import argv, exit
import struct
import qemu
import gdb

logger = logging.getLogger("TRACE")

## We need to keep some complex objects in between hook invocations so we keep them
## as globals. Tracing is inherently a single-threaded process, so using a
## manticore context would be heavier than needed.
stack_top = 0xC0000000
stack_size = 0x20000
initialized = False
last_instruction = None
in_helper = False


def init_logging():
    class ContextFilter(logging.Filter):
        def filter(self, record):
            record.stateid = ""
            return True

    logger.addFilter(ContextFilter())


def dump_gdb(cpu, addr, count):
    for offset in range(addr, addr + count, 4):
        val = int(gdb.getM(offset) & 0xFFFFFFFF)
        val2 = int(cpu.read_int(offset))
        print(f"{offset:x}: g{val:08x} m{val2:08x}")


def cmp_regs(cpu, should_print=False):
    """
    Compare registers from a remote gdb session to current mcore.

    :param manticore.core.cpu Cpu: Current cpu
    :param bool should_print: Whether to print values to stdout
    :return: Whether or not any differences were detected
    :rtype: bool
    """
    differing = False
    gdb_regs = gdb.getCanonicalRegisters()
    for name in sorted(gdb_regs):
        vg = gdb_regs[name]
        if name.endswith("psr"):
            name = "apsr"
        v = cpu.read_register(name.upper())
        if should_print:
            logger.debug(f"{name} gdb:{vg:x} mcore:{v:x}")
        if vg != v:
            if should_print:
                logger.warning("^^ unequal")
            differing = True
    if differing:
        logger.debug(qemu.correspond(None))
    return differing


def pre_mcore(state):
    # Start recording memory writes
    if state.cpu.instruction.mnemonic.lower() == "svc":
        state.cpu.memory.push_record_writes()


def post_mcore(state, last_instruction):
    """
    Handle syscalls (import memory) and bail if we diverge
    """
    global in_helper

    # Synchronize qemu state to manticore's after a system call
    if last_instruction.mnemonic.lower() == "svc":
        # Synchronize all writes that have happened
        writes = state.cpu.memory.pop_record_writes()
        if writes:
            logger.debug("Got %d writes", len(writes))
        for addr, val in writes:
            gdb.setByte(addr, val[0])

        # Write return val to gdb
        gdb_r0 = gdb.getR("R0")
        if gdb_r0 != state.cpu.R0:
            logger.debug(f"Writing 0x{state.cpu.R0:x} to R0 (overwriting 0x{gdb.getR('R0'):x})")
        for reg in state.cpu.canonical_registers:
            if reg.endswith("PSR") or reg in ("R15", "PC"):
                continue
            val = state.cpu.read_register(reg)
            gdb.setR(reg, val)

    # Ignore Linux kernel helpers
    if (state.cpu.PC >> 16) == 0xFFFF:
        in_helper = True
        return

    # If we executed a few instructions of a helper, we need to sync Manticore's
    # state to GDB as soon as we stop executing a helper.
    if in_helper:
        for reg in state.cpu.canonical_registers:
            if reg.endswith("PSR"):
                continue
            # Don't sync pc
            if reg == "R15":
                continue
            gdb.setR(reg, state.cpu.read_register(reg))
        in_helper = False

    if cmp_regs(state.cpu):
        cmp_regs(state.cpu, should_print=True)
        state.abandon()


def pre_qemu(state):
    # Nop for now, might need to do future sync state
    pass


def post_qemu(state, last_mnemonic):
    if last_mnemonic.lower() == "svc":
        sync_svc(state)


def sync_svc(state):
    """
    Mirror some service calls in manticore. Happens after qemu executed a SVC
    instruction, but before manticore did.
    """
    syscall = state.cpu.R7  # Grab idx from manticore since qemu could have exited
    name = linux_syscalls.armv7[syscall]

    logger.debug(f"Syncing syscall: {name}")

    try:
        # Make sure mmap returns the same address
        if "mmap" in name:
            returned = gdb.getR("R0")
            logger.debug(f"Syncing mmap ({returned:x})")
            state.cpu.write_register("R0", returned)
        if "exit" in name:
            return
    except ValueError:
        for reg in state.cpu.canonical_registers:
            print(f"{reg}: {state.cpu.read_register(reg):x}")
        raise


def initialize(state):
    """
    Synchronize the stack and register state (manticore->qemu)
    """
    logger.debug(f"Copying {stack_top - state.cpu.SP} bytes in the stack..")
    stack_bottom = min(state.cpu.SP, gdb.getR("SP"))
    for address in range(stack_bottom, stack_top):
        b = state.cpu.read_int(address, 8)
        gdb.setByte(address, chr(b))

    logger.debug("Done")

    # Qemu fd's start at 5, ours at 3. Add two filler fds
    mcore_stdout = state.platform.files[1]
    state.platform.files.append(mcore_stdout)
    state.platform.files.append(mcore_stdout)

    # Sync gdb's regs
    for gdb_reg in gdb.getCanonicalRegisters():
        if gdb_reg.endswith("psr"):
            mcore_reg = "APSR"
        else:
            mcore_reg = gdb_reg.upper()
        value = state.cpu.read_register(mcore_reg)
        gdb.setR(gdb_reg, value)


def verify(argv):
    logger.debug(f'Verifying program "{argv}"')

    # Address and stack_size are from linux.py
    # TODO(yan): Refactor these constants into a reachable value in platform
    qemu.start("arm", argv, va_size=stack_top, stack_size=stack_size)
    gdb.start("arm", argv)

    m = Manticore(argv[0], argv[1:])
    m.verbosity = 2

    init_logging()
    logger.setLevel(logging.DEBUG)

    @m.hook(None)
    def on_instruction(state):
        """
        Handle all the hooks for each instruction executed. Ordered as:

        pre_qemu
         * qemu exec *
        post_qemu

        // svc synchronization happens here (mmap specifically)

        pre_mcore
         * mcore exec *
        post_mcore

        // all memory written in a mcore syscall gets moved to qemu here
        """
        global initialized, last_instruction

        # Initialize our state to QEMU's
        if not initialized:
            initialize(state)
            initialized = True

        if last_instruction:
            post_mcore(state, last_instruction)

        # Kernel helpers are inline in QEMU; do nothing
        if (state.cpu.PC >> 16) == 0xFFFF:
            return

        pre_qemu(state)
        last_mnemonic = [x.strip() for x in gdb.getInstruction().split(":")][1].split("\t")[0]
        gdb.stepi()
        post_qemu(state, last_mnemonic)

        last_instruction = state.cpu.instruction
        pre_mcore(state)

    m.run()


if __name__ == "__main__":
    args = argv[1:]

    if len(args) == 0:
        print(f"usage: python {argv[0]} PROGRAM1 ...")
        exit()

    verify(args)