xemu-project/xemu
1
0
Fork 0
mirror of https://github.com/xemu-project/xemu.git synced 2024-11-25 04:30:02 +00:00
Code Issues Actions Packages Projects Releases Wiki Activity
cd093d5a2c
xemu/tests/image-fuzzer/runner.py
Eduardo Habkost 9fdd7860ad image-fuzzer: Use OSerror.strerror instead of tuple subscript 
OSError can't be used like a tuple on Python 3, so change the
code to use `e.sterror` instead of `e[1]`.

Reported-by: John Snow <jsnow@redhat.com>
Signed-off-by: Eduardo Habkost <ehabkost@redhat.com>
Reviewed-by: John Snow <jsnow@redhat.com>
Message-id: 20191021214117.18091-1-ehabkost@redhat.com
Message-Id: <20191021214117.18091-1-ehabkost@redhat.com>
Signed-off-by: Stefan Hajnoczi <stefanha@redhat.com>
2019-11-05 16:36:11 +01:00

432 lines
16 KiB
Python
Executable File
Raw Blame History

#!/usr/bin/env python3
# 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 time
import getopt
import io
import resource
try:
import json
except ImportError:
try:
import simplejson as json
except ImportError:
print("Warning: Module for JSON processing is not found.\n" \
"'--config' and '--command' options are not supported.", file=sys.stderr)
# 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.
"""
class Alarm(Exception):
"""Exception for signal.alarm events."""
pass
def handler(*args):
"""Notify that an alarm event occurred."""
raise Alarm
signal.signal(signal.SIGALRM, handler)
signal.alarm(600)
term_signal = signal.SIGKILL
devnull = open('/dev/null', 'r+')
process = subprocess.Popen(q_args, stdin=devnull,
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
errors='replace')
try:
out, err = process.communicate()
signal.alarm(0)
fd.write(out)
fd.write(err)
fd.flush()
return process.returncode
except Alarm:
os.kill(process.pid, term_signal)
fd.write('The command was terminated by timeout.\n')
fd.flush()
return -term_signal
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.maxsize))
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', '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 as e:
print("Error: The working directory '%s' cannot be used. Reason: %s"\
% (self.work_dir, e.strerror), file=sys.stderr)
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', '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 = io.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 = io.StringIO()
try:
retcode = run_app(temp_log, current_cmd)
except OSError as e:
multilog("%sError: Start of '%s' failed. Reason: %s\n\n"
% (test_summary, os.path.basename(current_cmd[0]),
e.strerror),
sys.stderr, self.log, self.parent_log)
raise TestException
if retcode < 0:
self.log.write(temp_log.getvalue())
multilog("%sFAIL: Test terminated by signal %s\n\n"
% (test_summary, 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("%sPASS: Application exited with the code " \
"'%d'\n\n" % (test_summary, 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
-d, --duration=NUMBER finish tests after NUMBER of seconds
-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 parameters 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()
def should_continue(duration, start_time):
"""Return True if a new test can be started and False otherwise."""
current_time = int(time.time())
return (duration is None) or (current_time - start_time < duration)
try:
opts, args = getopt.gnu_getopt(sys.argv[1:], 'c:hs:kvd:',
['command=', 'help', 'seed=', 'config=',
'keep_passed', 'verbose', 'duration='])
except getopt.error as e:
print("Error: %s\n\nTry 'runner.py --help' for more information" % e, file=sys.stderr)
sys.exit(1)
command = None
cleanup = True
log_all = False
seed = None
config = None
duration = 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) as e:
print("Error: JSON array of test commands cannot be loaded.\n" \
"Reason: %s" % e, file=sys.stderr)
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 in ('-d', '--duration'):
duration = int(arg)
elif opt == '--config':
try:
config = json.loads(arg)
except (TypeError, ValueError, NameError) as e:
print("Error: JSON array with the fuzzer configuration cannot" \
" be loaded\nReason: %s" % e, file=sys.stderr)
sys.exit(1)
if not len(args) == 2:
print("Expected two parameters\nTry 'runner.py --help'" \
" for more information.", file=sys.stderr)
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 as e:
print("Error: The image generator '%s' cannot be imported.\n" \
"Reason: %s" % (generator_name, e), file=sys.stderr)
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
start_time = int(time.time())
test_id = count(1)
while should_continue(duration, start_time):
try:
run_test(str(next(test_id)), seed, work_dir, run_log, cleanup,
log_all, command, config)
except (KeyboardInterrupt, SystemExit):
sys.exit(1)
if seed is not None:
break
Reference in New Issue View Git Blame Copy Permalink