path: root/tests/image-fuzzer/runner.py

#!/usr/bin/env python

# Tool for running fuzz tests
#
# Copyright (C) 2014 Maria Kustova <maria.k@catit.be>
#
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation, either version 2 of the License, or
# (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program.  If not, see <http://www.gnu.org/licenses/>.
#

import sys
import os
import signal
import subprocess
import random
import shutil
from itertools import count
import getopt
import StringIO
import resource

try:
    import json
except ImportError:
    try:
        import simplejson as json
    except ImportError:
        print >>sys.stderr, \
            "Warning: Module for JSON processing is not found.\n" \
            "'--config' and '--command' options are not supported."

# Backing file sizes in MB
MAX_BACKING_FILE_SIZE = 10
MIN_BACKING_FILE_SIZE = 1


def multilog(msg, *output):
    """ Write an object to all of specified file descriptors."""
    for fd in output:
        fd.write(msg)
        fd.flush()


def str_signal(sig):
    """ Convert a numeric value of a system signal to the string one
    defined by the current operational system.
    """
    for k, v in signal.__dict__.items():
        if v == sig:
            return k


def run_app(fd, q_args):
    """Start an application with specified arguments and return its exit code
    or kill signal depending on the result of execution.
    """
    devnull = open('/dev/null', 'r+')
    process = subprocess.Popen(q_args, stdin=devnull,
                               stdout=subprocess.PIPE,
                               stderr=subprocess.PIPE)
    out, err = process.communicate()
    fd.write(out)
    fd.write(err)
    return process.returncode


class TestException(Exception):
    """Exception for errors risen by TestEnv objects."""
    pass


class TestEnv(object):

    """Test object.

    The class sets up test environment, generates backing and test images
    and executes application under tests with specified arguments and a test
    image provided.

    All logs are collected.

    The summary log will contain short descriptions and statuses of tests in
    a run.

    The test log will include application (e.g. 'qemu-img') logs besides info
    sent to the summary log.
    """

    def __init__(self, test_id, seed, work_dir, run_log,
                 cleanup=True, log_all=False):
        """Set test environment in a specified work directory.

        Path to qemu-img and qemu-io will be retrieved from 'QEMU_IMG' and
        'QEMU_IO' environment variables.
        """
        if seed is not None:
            self.seed = seed
        else:
            self.seed = str(random.randint(0, sys.maxint))
        random.seed(self.seed)

        self.init_path = os.getcwd()
        self.work_dir = work_dir
        self.current_dir = os.path.join(work_dir, 'test-' + test_id)
        self.qemu_img = os.environ.get('QEMU_IMG', 'qemu-img')\
                                  .strip().split(' ')
        self.qemu_io = os.environ.get('QEMU_IO', 'qemu-io').strip().split(' ')
        self.commands = [['qemu-img', 'check', '-f', 'qcow2', '$test_img'],
                         ['qemu-img', 'info', '-f', 'qcow2', '$test_img'],
                         ['qemu-io', '$test_img', '-c', 'read $off $len'],
                         ['qemu-io', '$test_img', '-c', 'write $off $len'],
                         ['qemu-io', '$test_img', '-c',
                          'aio_read $off $len'],
                         ['qemu-io', '$test_img', '-c',
                          'aio_write $off $len'],
                         ['qemu-io', '$test_img', '-c', 'flush'],
                         ['qemu-io', '$test_img', '-c',
                          'discard $off $len'],
                         ['qemu-io', '$test_img', '-c',
                          'truncate $off']]
        for fmt in ['raw', 'vmdk', 'vdi', 'cow', 'qcow2', 'file',
                    'qed', 'vpc']:
            self.commands.append(
                ['qemu-img', 'convert', '-f', 'qcow2', '-O', fmt,
                 '$test_img', 'converted_image.' + fmt])

        try:
            os.makedirs(self.current_dir)
        except OSError, e:
            print >>sys.stderr, \
                "Error: The working directory '%s' cannot be used. Reason: %s"\
                % (self.work_dir, e[1])
            raise TestException
        self.log = open(os.path.join(self.current_dir, "test.log"), "w")
        self.parent_log = open(run_log, "a")
        self.failed = False
        self.cleanup = cleanup
        self.log_all = log_all

    def _create_backing_file(self):
        """Create a backing file in the current directory.

        Return a tuple of a backing file name and format.

        Format of a backing file is randomly chosen from all formats supported
        by 'qemu-img create'.
        """
        # All formats supported by the 'qemu-img create' command.
        backing_file_fmt = random.choice(['raw', 'vmdk', 'vdi', 'cow', 'qcow2',
                                          'file', 'qed', 'vpc'])
        backing_file_name = 'backing_img.' + backing_file_fmt
        backing_file_size = random.randint(MIN_BACKING_FILE_SIZE,
                                           MAX_BACKING_FILE_SIZE) * (1 << 20)
        cmd = self.qemu_img + ['create', '-f', backing_file_fmt,
                               backing_file_name, str(backing_file_size)]
        temp_log = StringIO.StringIO()
        retcode = run_app(temp_log, cmd)
        if retcode == 0:
            temp_log.close()
            return (backing_file_name, backing_file_fmt)
        else:
            multilog("Warning: The %s backing file was not created.\n\n"
                     % backing_file_fmt, sys.stderr, self.log, self.parent_log)
            self.log.write("Log for the failure:\n" + temp_log.getvalue() +
                           '\n\n')
            temp_log.close()
            return (None, None)

    def execute(self, input_commands=None, fuzz_config=None):
        """ Execute a test.

        The method creates backing and test images, runs test app and analyzes
        its exit status. If the application was killed by a signal, the test
        is marked as failed.
        """
        if input_commands is None:
            commands = self.commands
        else:
            commands = input_commands

        os.chdir(self.current_dir)
        backing_file_name, backing_file_fmt = self._create_backing_file()
        img_size = image_generator.create_image('test.img',
                                                backing_file_name,
                                                backing_file_fmt,
                                                fuzz_config)
        for item in commands:
            shutil.copy('test.img', 'copy.img')
            # 'off' and 'len' are multiple of the sector size
            sector_size = 512
            start = random.randrange(0, img_size + 1, sector_size)
            end = random.randrange(start, img_size + 1, sector_size)

            if item[0] == 'qemu-img':
                current_cmd = list(self.qemu_img)
            elif item[0] == 'qemu-io':
                current_cmd = list(self.qemu_io)
            else:
                multilog("Warning: test command '%s' is not defined.\n" \
                         % item[0], sys.stderr, self.log, self.parent_log)
                continue
            # Replace all placeholders with their real values
            for v in item[1:]:
                c = (v
                     .replace('$test_img', 'copy.img')
                     .replace('$off', str(start))
                     .replace('$len', str(end - start)))
                current_cmd.append(c)

            # Log string with the test header
            test_summary = "Seed: %s\nCommand: %s\nTest directory: %s\n" \
                           "Backing file: %s\n" \
                           % (self.seed, " ".join(current_cmd),
                              self.current_dir, backing_file_name)

            temp_log = StringIO.StringIO()
            try:
                retcode = run_app(temp_log, current_cmd)
            except OSError, e:
                multilog(test_summary + "Error: Start of '%s' failed. " \
                         "Reason: %s\n\n" % (os.path.basename(
                             current_cmd[0]), e[1]),
                         sys.stderr, self.log, self.parent_log)
                raise TestException

            if retcode < 0:
                self.log.write(temp_log.getvalue())
                multilog(test_summary + "FAIL: Test terminated by signal " +
                         "%s\n\n" % str_signal(-retcode), sys.stderr, self.log,
                         self.parent_log)
                self.failed = True
            else:
                if self.log_all:
                    self.log.write(temp_log.getvalue())
                    multilog(test_summary + "PASS: Application exited with" +
                             " the code '%d'\n\n" % retcode, sys.stdout,
                             self.log, self.parent_log)
            temp_log.close()
            os.remove('copy.img')

    def finish(self):
        """Restore the test environment after a test execution."""
        self.log.close()
        self.parent_log.close()
        os.chdir(self.init_path)
        if self.cleanup and not self.failed:
            shutil.rmtree(self.current_dir)

if __name__ == '__main__':

    def usage():
        print """
        Usage: runner.py [OPTION...] TEST_DIR IMG_GENERATOR

        Set up test environment in TEST_DIR and run a test in it. A module for
        test image generation should be specified via IMG_GENERATOR.
        Example:
        runner.py -c '[["qemu-img", "info", "$test_img"]]' /tmp/test qcow2

        Optional arguments:
          -h, --help                    display this help and exit
          -c, --command=JSON            run tests for all commands specified in
                                        the JSON array
          -s, --seed=STRING             seed for a test image generation,
                                        by default will be generated randomly
          --config=JSON                 take fuzzer configuration from the JSON
                                        array
          -k, --keep_passed             don't remove folders of passed tests
          -v, --verbose                 log information about passed tests

        JSON:

        '--command' accepts a JSON array of commands. Each command presents
        an application under test with all its paramaters as a list of strings,
        e.g.
          ["qemu-io", "$test_img", "-c", "write $off $len"]

        Supported application aliases: 'qemu-img' and 'qemu-io'.
        Supported argument aliases: $test_img for the fuzzed image, $off
        for an offset, $len for length.

        Values for $off and $len will be generated based on the virtual disk
        size of the fuzzed image
        Paths to 'qemu-img' and 'qemu-io' are retrevied from 'QEMU_IMG' and
        'QEMU_IO' environment variables

        '--config' accepts a JSON array of fields to be fuzzed, e.g.
          '[["header"], ["header", "version"]]'
        Each of the list elements can consist of a complex image element only
        as ["header"] or ["feature_name_table"] or an exact field as
        ["header", "version"]. In the first case random portion of the element
        fields will be fuzzed, in the second one the specified field will be
        fuzzed always.

        If '--config' argument is specified, fields not listed in
        the configuration array will not be fuzzed.
        """

    def run_test(test_id, seed, work_dir, run_log, cleanup, log_all,
                 command, fuzz_config):
        """Setup environment for one test and execute this test."""
        try:
            test = TestEnv(test_id, seed, work_dir, run_log, cleanup,
                           log_all)
        except TestException:
            sys.exit(1)

        # Python 2.4 doesn't support 'finally' and 'except' in the same 'try'
        # block
        try:
            try:
                test.execute(command, fuzz_config)
            except TestException:
                sys.exit(1)
        finally:
            test.finish()

    try:
        opts, args = getopt.gnu_getopt(sys.argv[1:], 'c:hs:kv',
                                       ['command=', 'help', 'seed=', 'config=',
                                        'keep_passed', 'verbose'])
    except getopt.error, e:
        print >>sys.stderr, \
            "Error: %s\n\nTry 'runner.py --help' for more information" % e
        sys.exit(1)

    command = None
    cleanup = True
    log_all = False
    seed = None
    config = None
    for opt, arg in opts:
        if opt in ('-h', '--help'):
            usage()
            sys.exit()
        elif opt in ('-c', '--command'):
            try:
                command = json.loads(arg)
            except (TypeError, ValueError, NameError), e:
                print >>sys.stderr, \
                    "Error: JSON array of test commands cannot be loaded.\n" \
                    "Reason: %s" % e
                sys.exit(1)
        elif opt in ('-k', '--keep_passed'):
            cleanup = False
        elif opt in ('-v', '--verbose'):
            log_all = True
        elif opt in ('-s', '--seed'):
            seed = arg
        elif opt == '--config':
            try:
                config = json.loads(arg)
            except (TypeError, ValueError, NameError), e:
                print >>sys.stderr, \
                    "Error: JSON array with the fuzzer configuration cannot" \
                    " be loaded\nReason: %s" % e
                sys.exit(1)

    if not len(args) == 2:
        print >>sys.stderr, \
            "Expected two parameters\nTry 'runner.py --help'" \
            " for more information."
        sys.exit(1)

    work_dir = os.path.realpath(args[0])
    # run_log is created in 'main', because multiple tests are expected to
    # log in it
    run_log = os.path.join(work_dir, 'run.log')

    # Add the path to the image generator module to sys.path
    sys.path.append(os.path.realpath(os.path.dirname(args[1])))
    # Remove a script extension from image generator module if any
    generator_name = os.path.splitext(os.path.basename(args[1]))[0]

    try:
        image_generator = __import__(generator_name)
    except ImportError, e:
        print >>sys.stderr, \
            "Error: The image generator '%s' cannot be imported.\n" \
            "Reason: %s" % (generator_name, e)
        sys.exit(1)

    # Enable core dumps
    resource.setrlimit(resource.RLIMIT_CORE, (-1, -1))
    # If a seed is specified, only one test will be executed.
    # Otherwise runner will terminate after a keyboard interruption
    for test_id in count(1):
        try:
            run_test(str(test_id), seed, work_dir, run_log, cleanup,
                     log_all, command, config)
        except (KeyboardInterrupt, SystemExit):
            sys.exit(1)

        if seed is not None:
            break