launcher_low_level.py

This shows how to connect to reven launcher in python, using reven low level api.

You will find some helpers to:

• Open/close a project.
• Generate a scenario from a binary.
• List available users, projects and virtual machines.

And a sample use case using them.

#!/usr/bin/env python2

import reven_api # to connect reven launcher
import reven # to give a high level handler on a reven server after opening a project

from os.path import basename
from time import sleep

def open_project(hostname, project_id):
    """
    Open a project and returns a reven.Project handler to it.
    @param hostname: hostname to join reven at
    @param project_id: reven_api.project_id
    @return a connected reven.Project or None
    """

    # connect to reven launcher
    lc = reven_api.launcher_connection(hostname)

    # check if project is already open
    port = lc.project_details(project_id).reven_server.port
    if port != 0:
        print "Project already running on port %s" % port
        return reven.Project(hostname, port)

    # actually open project
    port = lc.server_launch(project_id)
    if port == 0:
        print "Server not started"
        return None

    print "Server started on port %s" % port

    # reven server cannnot accept connections immediatly after beeing launched
    # try to reconnect if first attempts failed
    for i in range(10):
        try:
            client = reven.Project(hostname, port)
            print "Connected!"
            return client
        except RuntimeError as e:
            print e
            sleep(1)
            pass

def close_project(hostname, project_id):
    """
    Close a project (kill its running server).
    @param hostname: hostname to join reven at
    @param project_id: reven_api.project_id
    """

    # connect to reven launcher
    lc = reven_api.launcher_connection(hostname)

    # close project
    port = lc.project_details(project_id).reven_server.port
    lc.server_kill(port)

    # check project is properly closed
    if lc.project_details(project_id).reven_server.port:
        print "Server not killed"
    else:
        print "Project closed"


def list_vms(lc):
    """
    Print configuration of all available virtual machines.
    @param lc: reven_api.launcher_connection
    """

    print "Available virtual machines:"

    for vm in lc.list_vms():
        # here vm is a reven_api.vm_info
        print "[%s]" % vm.name
        print "vbox_name = %s" % vm.vbox_name
        print "display = %s" % vm.display

        print "dynamic_launch = %s" % vm.dynamic_launch
        print "static_launch = %s" % vm.static_launch
        print "stopper = %s" % vm.stopper

        print "os = %s" % vm.os
        print "segment = %s" % vm.segment
        print "pdb_path = %s" % vm.pdb_path

        print "vnc_password = %s" % vm.vnc_password
        print "vnc_port = %s" % vm.vnc_port
        print ""

def list_users(lc):
    """
    List all reven users.
    @param lc: reven_api.launcher_connection
    """

    print "Reven users:"

    for user in lc.list_users():
        # here user is a string
        print user

def list_projects_for_user(lc, user):
    """
    List all project for given user.
    @param lc: reven_api.launcher_connection
    @param user: username as string
    """

    print "Projects for user %s:" % user

    for project in lc.list_projects(user):
        # here project is a reven_api.project_id to be forwarded to launcher_connection services call
        print "%s/%s" % (project.user, project.project)

def wait_end_of_scenario(lc, project_id):
    """
    Block until end of scenario generation.
    @param lc: reven_api.launcher_connection
    @param project_id: reven_api.project_id
    @return True if generation is successful
    """

    while True:
        progress = lc.project_scenario(project_id)
        print progress.log_chunk
        if not progress.is_generating:
            break
        # could abort scenario generation not to wait forever with:
        # lc.project_stop_scenario_generation(project_id)
        sleep(1)

    return progress.is_successful

def generate_scenario(lc, project_id, vm, binary, args='', input_files=[], dump_at=''):
    """
    Generate a scenario
    @param lc: reven_api.launcher_connection
    @param project_id: reven.project_id
    @param binary: binary filepath
    @param args: arguments string for binary launch
    @param input_files: list of additionnal input files
    @param dump_at: symbol name or address in binary, start scenario generation when reached
    """

    # upload all input files, including binary
    binary_name = basename(binary)
    if not binary in input_files:
        input_files.append(binary)

    for filepath in input_files:
        lc.project_upload_file(project_id, filepath)

    # configure scenario
    sc = reven_api.scenario_config()
    sc.vm_config_name = vm
    sc.binary_name = binary_name
    sc.binary_arguments = args or ''
    sc.binary_dump_hint = dump_at or ''
    sc.binary_dump_address = dump_at or ''
    sc.system_pdb_path = ''

    # configure generation
    sglc = reven_api.scenario_generation_launch_config()
    sglc.enable_instruction_tracing = False
    sglc.is_interactive = False
    sglc.vnc_password = ''
    sglc.vnc_port = ''

    scenario_config = reven_api.scenario_generation_config()
    scenario_config.generation = sglc
    scenario_config.scenario = sc

    print "Start scenario generation"
    lc.project_generate_scenario(project_id, scenario_config)

    if wait_end_of_scenario(lc, project_id):
        print "Generation ok"
    else:
        print "Generation failed"


if __name__ == '__main__':
    # example config:
    launcher_hostname = 'localhost'
    sample_user = 'reven'
    sample_project = reven_api.project_id(sample_user, '_sample')
    sample_binary = '/tmp/lynx32'
    sample_binary_arguments = ' -dump crash.html'
    sample_input_files = ['/tmp/crash.html']
    sample_vm = 'vm_debian_auto'

    lc = reven_api.launcher_connection(launcher_hostname)

    # list info from launcher
    list_vms(lc)
    list_users(lc)
    print ''
    list_projects_for_user(lc, lc.list_users()[0])

    # create project
    lc.project_create(sample_project)

    # if previous execution of this script was interrupted:
    # - close project if running
    close_project(launcher_hostname, sample_project)
    # - stop previous scenario generation, wait for it to end
    lc.project_stop_scenario_generation(sample_project)
    wait_end_of_scenario(lc, sample_project)

    # generate scenario
    generate_scenario(lc, sample_project, sample_vm, sample_binary, args=sample_binary_arguments, input_files=sample_input_files)

    # open project
    client = open_project(launcher_hostname, sample_project)

    # start and monitor execution
    client.start_execution()
    while True:
        progress = client.execution_status()

        print "%s/%s - %s/%s - %s" % (
            progress.tsc, progress.last_tsc,
            progress.point_index, progress.last_point_index,
            progress.status)

        if not progress.is_busy:
            break

        sleep(1)

    # use reven
    print client.traces()

    if list(client.traces()[0].search_symbol('my_awesome_symbol')):
        print "Symbol found, could keep project"
        client.save_execution('good')
    else:
        print "Symbol not found, could discard project"

    # close project
    close_project(launcher_hostname, sample_project)

    # delete sample project
    lc.project_delete(sample_project)