smc_check.py

• Linux path to script:
  • /usr/share/reven/examples/smc_check.py
• Windows path to script:
  • amd64: C:\Program Files (x86)\Reven-PythonAPI-amd64\python-examples\smc_check.py
  • x86: C:\Program Files (x86)\Reven-PythonAPI-x86\python-examples\smc_check.py

This example demonstrates how to use the python API to find instructions in a given process that were generated at runtime (execution-after-write).

See the script's documentation below for more information.

Note that this script only iterates over points that belong to a given process, and not over the whole trace. For more details, look at the implementation of the get_process_point_ranges and get_process_sequence_points methods.

"""
Purpose:
  This script shows the runtime generated instructions of a given process

  It is similar in spirit to the `executions-after-write` inspector, but done at a process level rather than on the
  whole trace, and directly from python after the trace execution (the `executions-after-write` inspector is
  not required for this script to work).

Usage:
  Give the following arguments:
    - host:port of the REVEN's project (default: localhost:13370)
    - process CR3 or PID.
"""


import reven
import sys
import argparse


def check_process(project, cr3):
    """
    Check if a process is running in the whole trace.
    @param cr3: The CR3 register's value associated to the process to check.
    """
    for process in project.processes():
        if process.cr3 == cr3:
            return
    raise ValueError('Unknown process with cr3 = {}'.format(hex(cr3)))

def get_cr3_from_pid(project, pid):
    """
    Get the CR3 register's value associated to a running process.
    @param pid: The PID of the process.
    """
    for process in project.processes():
        if process.pid == pid:
            return process.cr3
    raise ValueError('Unknown process with pid = {}'.format(pid))

def get_last_point(trace):
    """
    Get the last point of the given trace.
    """
    last_sequence = trace.sequence_count - 1
    last_instruction = trace.sequence_length(last_sequence) - 1
    return trace.point(last_sequence, last_instruction)

def get_process_point_ranges(project, cr3):
    """
    Get the ranges of main trace (e.g. 'Execution run') points where the process is running.
    @param cr3: The CR3 register's value.
    @return: A generator of tuples composed of the start and end point([start, end]).
    """
    main_trace = project.traces()[0] # Execution run
    first_point = main_trace.point(0, 0)
    start = first_point if first_point.cpu().read_register('cr3') == cr3 else None
    for switch in project.process_switches():
        if main_trace.name != switch.point.trace.name:
            continue
        if start is None:
            if cr3 == switch.cr3:
                start = switch.point.next_sequence()
        else:
            yield (start, switch.point)
            start = None
    if start:
        # special case where the trace finishes while the process is running.
        yield (start, get_last_point(main_trace))

def get_process_sequence_points(project, cr3):
    """
    Get the first points of main trace (e.g. 'Execution run') sequences where the process is running in the main trace.
    @param cr3: The CR3 register's value.
    @return: A generator of points belonging to the main trace.
    """
    for first, last in get_process_point_ranges(project, cr3):
        point = first
        while point is not None and point <= last:
            yield point
            point = point.next_sequence()

class EmptySequencePoints(RuntimeError):
    pass

def check_smc(sequence_points):
    accessed_memory_type = {}
    smc_addresses = set()
    has_sequence_points = False

    # subtyping for accessed memory: W<:X
    # casting
    #   up   X <== W is allowed
    #   down X ==> W is not
    def update_memory_type(address, mem_type):
        if (address in accessed_memory_type) and \
           (accessed_memory_type[address] == 'W') and \
           (mem_type == 'X'):
            smc_addresses.add(address)
            raise TypeError(hex(address))
        else:
            accessed_memory_type[address] = mem_type

    for sequence_point in sequence_points:
        has_sequence_points = True
        point = sequence_point
        basic_block = point.basic_block
        for instruction in basic_block:
            ins_addr = instruction.address
            ins_len = instruction.size
            for addr in range(ins_addr, ins_addr + ins_len):
                try:
                    update_memory_type(addr, 'X')
                except TypeError as e:
                    print('Self-modifying code detected: {} (X ==> W)').format(e)

            mem_accesses = point.memory_accesses()
            for mem_access in mem_accesses:
                if mem_access.is_write:
                    for addr in range(mem_access.address, mem_access.address + mem_access.size):
                        update_memory_type(addr, 'W')

            point = point.next()

    if not has_sequence_points:
        raise EmptySequencePoints()

    return smc_addresses

def main(project, cr3, pid):
    if cr3 is not None:
        check_process(project, cr3)
    else:
        cr3 = get_cr3_from_pid(project, pid)
    sequence_points = get_process_sequence_points(project, cr3)

    try:
        smc_addresses = check_smc(sequence_points)
    except EmptySequencePoints:
        print('The process is not running in the main trace.')
        return

    if not smc_addresses:
        print("Self-modifying code not found")

def auto_int(x):
    return int(x, 0)

if __name__ == '__main__':
    parser = argparse.ArgumentParser()
    parser.add_argument("--host", type=str, help='The reven server host. (default = "localhost")', default="localhost")
    parser.add_argument("--port", type=int, help="The reven server port. (default = 13370)", default=13370)
    subparsers = parser.add_subparsers(dest="subparser", help="Process identification")
    cr3_parser = subparsers.add_parser('cr3', help="Use the CR3 register to identify a process (recommended).")
    cr3_parser.add_argument("value", type=auto_int, help="The CR3 register value")
    pid_parser = subparsers.add_parser('pid', help="Use the PID to identify a process.")
    pid_parser.add_argument("value", type=int, help="The PID value")
    args = parser.parse_args()

    cr3 = args.value if args.subparser == 'cr3' else None
    pid = args.value if args.subparser == 'pid' else None

    try:
        project = reven.Project(args.host, args.port)
        main(project, cr3, pid)
    except (RuntimeError, ValueError) as e:
        print('Error: {}').format(e)
    except:
        print('Unknown error: {}').format(sys.exc_info())