percent.py

The percent script simulates the % key in VI command mode, more stack memory items. Basically this means that for a push, this makes you jump to the associated pop. For a call, this makes you jump to the corresponding ret, and so on.

NOTE: This example make use of the low-level api.

This will demonstrate the use of the following services:

• run_get_running_context_between
• memory_get_physical_address
• memory_get_history_between
• run_search_next_memory_use
• run_search_next_register_use

It is built as an Axion script, so this also shows the use of a callback in Axion.

from axion_api import axion
import reven_api
from PythonQt import Qt


def register_command(callback, desc, name, shortcut):
    action = Qt.QAction(desc, axion)
    axion.shortcuts().register_action(action, name, Qt.QKeySequence(shortcut))
    action.connect("triggered(bool)", callback)

def reven_connection_from_axion(axion):
    host, port = axion.connection_info()
    if port == 0 or not host:
        return None
    return reven_api.reven_connection(host.encode(), port)

def is_memory_history_enabled(client):
    return 'memory_range_history' in [ i.name for i in client.engine_get_current_preset() if i.properties.enabled ]


def pick_memory_access(client, point):
    "Return first non null logical address accessed by instruction at given execution point or None"

    null_logical = reven_api.logical_address(0, 0)
    result = None

    accesses = client.memory_get_history_instruction(point)
    for access in accesses:
        if access.logical != null_logical:
            if not (result and result.write and not access.write):
                result = access
    return result

def get_matching_instruction(client, point):
    "Return execution point of matching instruction for given execution point or None if not found."

    # create a range of size 1 to query context before and after selected instruction
    point_range = reven_api.execution_range(point.run_name, point.sequence_identifier, 1, point.instruction_index)
    # give an empty vector of logical address range to ignore memory for we don't nedd any memory here
    context = client.run_get_running_context_between(point_range, reven_api.vector_of_logical_address_range())

    # ss values before and after selected instruction
    ss_before = context.before.numeric_registers['ss'].value
    ss_after = context.after.numeric_registers['ss'].value

    # ss basic heuristic to handle sysenter/sysexit
    if ss_before != ss_after:
        # ss is modified by instruction, search next or previous write for ss and return corresponding execution point
        return client.run_search_next_register_use(point, forward=(ss_before > ss_after), read=False, write=True, register_name="ss")

    # memory heuristic
    access = pick_memory_access(client, point)
    if access != None:
        if access.write:
            return client.run_search_next_memory_use(point, forward=True, read=True, write=False, address=access.logical, size=access.size)

        if access.read:
            return client.run_search_next_memory_use(point, forward=False, read=False, write=True, address=access.logical, size=access.size)

    # esp values before and after selected instruction
    esp_before = context.before.numeric_registers['esp'].value
    esp_after = context.after.numeric_registers['esp'].value

    # esp heuristic
    if esp_before != esp_after:
        if esp_before > esp_after:
            stack = reven_api.logical_address(ss_after, esp_after)
            return client.run_search_next_memory_use(point, forward=True, read=True, write=False, address=stack, size=4)

        if esp_before < esp_after:
            stack = reven_api.logical_address(ss_before, esp_before)
            return client.run_search_next_memory_use(point, forward=False, read=False, write=True, address=stack, size=4)

def trigger_percent():
    "Callback to call when this plugin is enabled."

    # get current selected instruction from axion
    run, seq, instr = axion.selected_sequence()

    # connect to Reven server project instance for latter service calls
    client = reven_connection_from_axion(axion)

    # get reven execution point corresponding to selected instruction (for reven service calls take execution point)
    point = reven_api.execution_point(run.encode(), seq, instr)

    # retrieve data from Reven and use custom heuristic to find an instruction matching the current one
    result = get_matching_instruction(client, point)

    if result == None or not result.valid():
        axion.status_message("No matching instruction recorded for [%s@%d:%d]" % (run, seq, instr))
    else:
         # select matching instruction in axion
        axion.select_sequence(result.run_name, result.sequence_identifier, result.instruction_index)
        axion.status_message("Jumped to matching instruction at [%s@%d:%d] from [%s@%d:%d]" % (result.run_name, result.sequence_identifier, result.instruction_index, run, seq, instr))


def get_disabled_callback(reason):
    "Generate a callback printing reason for having disabled this plugin in status bar"

    def disabled():
        axion.status_message("Percent plugin is disabled (%s)" % reason)

    return disabled

def try_enable():
    "Try to enable this plugin, check connection and memory history availability"

    global current_callback

    client = reven_connection_from_axion(axion)
    if not client:
        current_callback = get_disabled_callback("Not connected to Reven server")
        return

    if not is_memory_history_enabled(client) :
        reason = "requires inspector memory range history"
        current_callback = get_disabled_callback(reason)
        axion.log_warning("Percent plugin disabled (%s)" % reason)
    else:
        current_callback = trigger_percent


# DO NOT simply assign axion_callback to callback_enabled/disabled (would break with register_command's call)
def axion_callback():
    "Actual callback auto-called by axion when this plugin is triggered"

    current_callback()

def axion_connected():
    "Callback auto-called by axion when connected to a new server"

    try_enable()

def axion_disconnected():
    "Callback auto-called by axion when disconnected from current server"

    current_callback = get_disabled_callback("Not connected to Reven server")


current_callback = get_disabled_callback("Not initialised")
register_command(axion_callback, "percent", "plugin.percent", "%")
try_enable()