diff options
Diffstat (limited to 'scripts/oss-fuzz')
| -rwxr-xr-x | scripts/oss-fuzz/build.sh | 115 | ||||
| -rw-r--r-- | scripts/oss-fuzz/instrumentation-filter-template | 15 | ||||
| -rwxr-xr-x | scripts/oss-fuzz/minimize_qtest_trace.py | 323 | ||||
| -rwxr-xr-x | scripts/oss-fuzz/output_reproducer.py | 160 | ||||
| -rwxr-xr-x | scripts/oss-fuzz/reorder_fuzzer_qtest_trace.py | 103 |
5 files changed, 716 insertions, 0 deletions
diff --git a/scripts/oss-fuzz/build.sh b/scripts/oss-fuzz/build.sh new file mode 100755 index 000000000..98b56e052 --- /dev/null +++ b/scripts/oss-fuzz/build.sh @@ -0,0 +1,115 @@ +#!/bin/sh -e +# +# OSS-Fuzz build script. See: +# https://google.github.io/oss-fuzz/getting-started/new-project-guide/#buildsh +# +# The file is consumed by: +# https://github.com/google/oss-fuzz/blob/master/projects/qemu/Dockerfiles +# +# This code is licensed under the GPL version 2 or later. See +# the COPYING file in the top-level directory. +# + +# build project +# e.g. +# ./autogen.sh +# ./configure +# make -j$(nproc) all + +# build fuzzers +# e.g. +# $CXX $CXXFLAGS -std=c++11 -Iinclude \ +# /path/to/name_of_fuzzer.cc -o $OUT/name_of_fuzzer \ +# -fsanitize=fuzzer /path/to/library.a + +fatal () { + echo "Error : ${*}, exiting." + exit 1 +} + +OSS_FUZZ_BUILD_DIR="./build-oss-fuzz/" + +# There seems to be a bug in clang-11 (used for builds on oss-fuzz) : +# accel/tcg/cputlb.o: In function `load_memop': +# accel/tcg/cputlb.c:1505: undefined reference to `qemu_build_not_reached' +# +# When building with optimization, the compiler is expected to prove that the +# statement cannot be reached, and remove it. For some reason clang-11 doesn't +# remove it, resulting in an unresolved reference to qemu_build_not_reached +# Undefine the __OPTIMIZE__ macro which compiler.h relies on to choose whether +# to " #define qemu_build_not_reached() g_assert_not_reached() " +EXTRA_CFLAGS="$CFLAGS -U __OPTIMIZE__" + +if ! { [ -e "./COPYING" ] && + [ -e "./MAINTAINERS" ] && + [ -e "./Makefile" ] && + [ -e "./docs" ] && + [ -e "./VERSION" ] && + [ -e "./linux-user" ] && + [ -e "./softmmu" ];} ; then + fatal "Please run the script from the top of the QEMU tree" +fi + +mkdir -p $OSS_FUZZ_BUILD_DIR || fatal "mkdir $OSS_FUZZ_BUILD_DIR failed" +cd $OSS_FUZZ_BUILD_DIR || fatal "cd $OSS_FUZZ_BUILD_DIR failed" + + +if [ -z ${OUT+x} ]; then + DEST_DIR=$(realpath "./DEST_DIR") +else + DEST_DIR=$OUT +fi + +mkdir -p "$DEST_DIR/lib/" # Copy the shared libraries here + +# Build once to get the list of dynamic lib paths, and copy them over +../configure --disable-werror --cc="$CC" --cxx="$CXX" --enable-fuzzing \ + --prefix="$DEST_DIR" --bindir="$DEST_DIR" --datadir="$DEST_DIR/data/" \ + --extra-cflags="$EXTRA_CFLAGS" --target-list="i386-softmmu" + +if ! make "-j$(nproc)" qemu-fuzz-i386; then + fatal "Build failed. Please specify a compiler with fuzzing support"\ + "using the \$CC and \$CXX environment variables"\ + "\nFor example: CC=clang CXX=clang++ $0" +fi + +if [ "$GITLAB_CI" != "true" ]; then + for i in $(ldd ./qemu-fuzz-i386 | cut -f3 -d' '); do + cp "$i" "$DEST_DIR/lib/" + done + rm qemu-fuzz-i386 + + # Build a second time to build the final binary with correct rpath + ../configure --disable-werror --cc="$CC" --cxx="$CXX" --enable-fuzzing \ + --prefix="$DEST_DIR" --bindir="$DEST_DIR" --datadir="$DEST_DIR/data/" \ + --extra-cflags="$EXTRA_CFLAGS" --extra-ldflags="-Wl,-rpath,\$ORIGIN/lib" \ + --target-list="i386-softmmu" + make "-j$(nproc)" qemu-fuzz-i386 V=1 +fi + +# Copy over the datadir +cp -r ../pc-bios/ "$DEST_DIR/pc-bios" + +targets=$(./qemu-fuzz-i386 | awk '$1 ~ /\*/ {print $2}') +base_copy="$DEST_DIR/qemu-fuzz-i386-target-$(echo "$targets" | head -n 1)" + +cp "./qemu-fuzz-i386" "$base_copy" + +# Run the fuzzer with no arguments, to print the help-string and get the list +# of available fuzz-targets. Copy over the qemu-fuzz-i386, naming it according +# to each available fuzz target (See 05509c8e6d fuzz: select fuzz target using +# executable name) +for target in $(echo "$targets" | tail -n +2); +do + # Ignore the generic-fuzz target, as it requires some environment variables + # to be configured. We have some generic-fuzz-{pc-q35, floppy, ...} targets + # that are thin wrappers around this target that set the required + # environment variables according to predefined configs. + if [ "$target" != "generic-fuzz" ]; then + ln $base_copy \ + "$DEST_DIR/qemu-fuzz-i386-target-$target" + fi +done + +echo "Done. The fuzzers are located in $DEST_DIR" +exit 0 diff --git a/scripts/oss-fuzz/instrumentation-filter-template b/scripts/oss-fuzz/instrumentation-filter-template new file mode 100644 index 000000000..76d2b6139 --- /dev/null +++ b/scripts/oss-fuzz/instrumentation-filter-template @@ -0,0 +1,15 @@ +# Code that we actually want the fuzzer to target +# See: https://clang.llvm.org/docs/SanitizerCoverage.html#disabling-instrumentation-without-source-modification +# +src:*/hw/* +src:*/include/hw/* +src:*/slirp/* +src:*/net/* + +# We don't care about coverage over fuzzer-specific code, however we should +# instrument the fuzzer entry-point so libFuzzer always sees at least some +# coverage - otherwise it will exit after the first input +src:*/tests/qtest/fuzz/fuzz.c + +# Enable instrumentation for all functions in those files +fun:* diff --git a/scripts/oss-fuzz/minimize_qtest_trace.py b/scripts/oss-fuzz/minimize_qtest_trace.py new file mode 100755 index 000000000..20825768c --- /dev/null +++ b/scripts/oss-fuzz/minimize_qtest_trace.py @@ -0,0 +1,323 @@ +#!/usr/bin/env python3 +# -*- coding: utf-8 -*- + +""" +This takes a crashing qtest trace and tries to remove superflous operations +""" + +import sys +import os +import subprocess +import time +import struct + +QEMU_ARGS = None +QEMU_PATH = None +TIMEOUT = 5 +CRASH_TOKEN = None + +# Minimization levels +M1 = False # try removing IO commands iteratively +M2 = False # try setting bits in operand of write/out to zero + +write_suffix_lookup = {"b": (1, "B"), + "w": (2, "H"), + "l": (4, "L"), + "q": (8, "Q")} + +def usage(): + sys.exit("""\ +Usage: + +QEMU_PATH="/path/to/qemu" QEMU_ARGS="args" {} [Options] input_trace output_trace + +By default, will try to use the second-to-last line in the output to identify +whether the crash occred. Optionally, manually set a string that idenitifes the +crash by setting CRASH_TOKEN= + +Options: + +-M1: enable a loop around the remove minimizer, which may help decrease some + timing dependant instructions. Off by default. +-M2: try setting bits in operand of write/out to zero. Off by default. + +""".format((sys.argv[0]))) + +deduplication_note = """\n\ +Note: While trimming the input, sometimes the mutated trace triggers a different +type crash but indicates the same bug. Under this situation, our minimizer is +incapable of recognizing and stopped from removing it. In the future, we may +use a more sophisticated crash case deduplication method. +\n""" + +def check_if_trace_crashes(trace, path): + with open(path, "w") as tracefile: + tracefile.write("".join(trace)) + + rc = subprocess.Popen("timeout -s 9 {timeout}s {qemu_path} {qemu_args} 2>&1\ + < {trace_path}".format(timeout=TIMEOUT, + qemu_path=QEMU_PATH, + qemu_args=QEMU_ARGS, + trace_path=path), + shell=True, + stdin=subprocess.PIPE, + stdout=subprocess.PIPE, + encoding="utf-8") + global CRASH_TOKEN + if CRASH_TOKEN is None: + try: + outs, _ = rc.communicate(timeout=5) + CRASH_TOKEN = " ".join(outs.splitlines()[-2].split()[0:3]) + except subprocess.TimeoutExpired: + print("subprocess.TimeoutExpired") + return False + print("Identifying Crashes by this string: {}".format(CRASH_TOKEN)) + global deduplication_note + print(deduplication_note) + return True + + for line in iter(rc.stdout.readline, ""): + if "CLOSED" in line: + return False + if CRASH_TOKEN in line: + return True + + print("\nWarning:") + print(" There is no 'CLOSED'or CRASH_TOKEN in the stdout of subprocess.") + print(" Usually this indicates a different type of crash.\n") + return False + + +# If previous write commands write the same length of data at the same +# interval, we view it as a hint. +def split_write_hint(newtrace, i): + HINT_LEN = 3 # > 2 + if i <=(HINT_LEN-1): + return None + + #find previous continuous write traces + k = 0 + l = i-1 + writes = [] + while (k != HINT_LEN and l >= 0): + if newtrace[l].startswith("write "): + writes.append(newtrace[l]) + k += 1 + l -= 1 + elif newtrace[l] == "": + l -= 1 + else: + return None + if k != HINT_LEN: + return None + + length = int(writes[0].split()[2], 16) + for j in range(1, HINT_LEN): + if length != int(writes[j].split()[2], 16): + return None + + step = int(writes[0].split()[1], 16) - int(writes[1].split()[1], 16) + for j in range(1, HINT_LEN-1): + if step != int(writes[j].split()[1], 16) - \ + int(writes[j+1].split()[1], 16): + return None + + return (int(writes[0].split()[1], 16)+step, length) + + +def remove_lines(newtrace, outpath): + remove_step = 1 + i = 0 + while i < len(newtrace): + # 1.) Try to remove lines completely and reproduce the crash. + # If it works, we're done. + if (i+remove_step) >= len(newtrace): + remove_step = 1 + prior = newtrace[i:i+remove_step] + for j in range(i, i+remove_step): + newtrace[j] = "" + print("Removing {lines} ...\n".format(lines=prior)) + if check_if_trace_crashes(newtrace, outpath): + i += remove_step + # Double the number of lines to remove for next round + remove_step *= 2 + continue + # Failed to remove multiple IOs, fast recovery + if remove_step > 1: + for j in range(i, i+remove_step): + newtrace[j] = prior[j-i] + remove_step = 1 + continue + newtrace[i] = prior[0] # remove_step = 1 + + # 2.) Try to replace write{bwlq} commands with a write addr, len + # command. Since this can require swapping endianness, try both LE and + # BE options. We do this, so we can "trim" the writes in (3) + + if (newtrace[i].startswith("write") and not + newtrace[i].startswith("write ")): + suffix = newtrace[i].split()[0][-1] + assert(suffix in write_suffix_lookup) + addr = int(newtrace[i].split()[1], 16) + value = int(newtrace[i].split()[2], 16) + for endianness in ['<', '>']: + data = struct.pack("{end}{size}".format(end=endianness, + size=write_suffix_lookup[suffix][1]), + value) + newtrace[i] = "write {addr} {size} 0x{data}\n".format( + addr=hex(addr), + size=hex(write_suffix_lookup[suffix][0]), + data=data.hex()) + if(check_if_trace_crashes(newtrace, outpath)): + break + else: + newtrace[i] = prior[0] + + # 3.) If it is a qtest write command: write addr len data, try to split + # it into two separate write commands. If splitting the data operand + # from length/2^n bytes to the left does not work, try to move the pivot + # to the right side, then add one to n, until length/2^n == 0. The idea + # is to prune unneccessary bytes from long writes, while accommodating + # arbitrary MemoryRegion access sizes and alignments. + + # This algorithm will fail under some rare situations. + # e.g., xxxxxxxxxuxxxxxx (u is the unnecessary byte) + + if newtrace[i].startswith("write "): + addr = int(newtrace[i].split()[1], 16) + length = int(newtrace[i].split()[2], 16) + data = newtrace[i].split()[3][2:] + if length > 1: + + # Can we get a hint from previous writes? + hint = split_write_hint(newtrace, i) + if hint is not None: + hint_addr = hint[0] + hint_len = hint[1] + if hint_addr >= addr and hint_addr+hint_len <= addr+length: + newtrace[i] = "write {addr} {size} 0x{data}\n".format( + addr=hex(hint_addr), + size=hex(hint_len), + data=data[(hint_addr-addr)*2:\ + (hint_addr-addr)*2+hint_len*2]) + if check_if_trace_crashes(newtrace, outpath): + # next round + i += 1 + continue + newtrace[i] = prior[0] + + # Try splitting it using a binary approach + leftlength = int(length/2) + rightlength = length - leftlength + newtrace.insert(i+1, "") + power = 1 + while leftlength > 0: + newtrace[i] = "write {addr} {size} 0x{data}\n".format( + addr=hex(addr), + size=hex(leftlength), + data=data[:leftlength*2]) + newtrace[i+1] = "write {addr} {size} 0x{data}\n".format( + addr=hex(addr+leftlength), + size=hex(rightlength), + data=data[leftlength*2:]) + if check_if_trace_crashes(newtrace, outpath): + break + # move the pivot to right side + if leftlength < rightlength: + rightlength, leftlength = leftlength, rightlength + continue + power += 1 + leftlength = int(length/pow(2, power)) + rightlength = length - leftlength + if check_if_trace_crashes(newtrace, outpath): + i -= 1 + else: + newtrace[i] = prior[0] + del newtrace[i+1] + i += 1 + + +def clear_bits(newtrace, outpath): + # try setting bits in operands of out/write to zero + i = 0 + while i < len(newtrace): + if (not newtrace[i].startswith("write ") and not + newtrace[i].startswith("out")): + i += 1 + continue + # write ADDR SIZE DATA + # outx ADDR VALUE + print("\nzero setting bits: {}".format(newtrace[i])) + + prefix = " ".join(newtrace[i].split()[:-1]) + data = newtrace[i].split()[-1] + data_bin = bin(int(data, 16)) + data_bin_list = list(data_bin) + + for j in range(2, len(data_bin_list)): + prior = newtrace[i] + if (data_bin_list[j] == '1'): + data_bin_list[j] = '0' + data_try = hex(int("".join(data_bin_list), 2)) + # It seems qtest only accepts padded hex-values. + if len(data_try) % 2 == 1: + data_try = data_try[:2] + "0" + data_try[2:] + + newtrace[i] = "{prefix} {data_try}\n".format( + prefix=prefix, + data_try=data_try) + + if not check_if_trace_crashes(newtrace, outpath): + data_bin_list[j] = '1' + newtrace[i] = prior + i += 1 + + +def minimize_trace(inpath, outpath): + global TIMEOUT + with open(inpath) as f: + trace = f.readlines() + start = time.time() + if not check_if_trace_crashes(trace, outpath): + sys.exit("The input qtest trace didn't cause a crash...") + end = time.time() + print("Crashed in {} seconds".format(end-start)) + TIMEOUT = (end-start)*5 + print("Setting the timeout for {} seconds".format(TIMEOUT)) + + newtrace = trace[:] + global M1, M2 + + # remove lines + old_len = len(newtrace) + 1 + while(old_len > len(newtrace)): + old_len = len(newtrace) + print("trace lenth = ", old_len) + remove_lines(newtrace, outpath) + if not M1 and not M2: + break + newtrace = list(filter(lambda s: s != "", newtrace)) + assert(check_if_trace_crashes(newtrace, outpath)) + + # set bits to zero + if M2: + clear_bits(newtrace, outpath) + assert(check_if_trace_crashes(newtrace, outpath)) + + +if __name__ == '__main__': + if len(sys.argv) < 3: + usage() + if "-M1" in sys.argv: + M1 = True + if "-M2" in sys.argv: + M2 = True + QEMU_PATH = os.getenv("QEMU_PATH") + QEMU_ARGS = os.getenv("QEMU_ARGS") + if QEMU_PATH is None or QEMU_ARGS is None: + usage() + # if "accel" not in QEMU_ARGS: + # QEMU_ARGS += " -accel qtest" + CRASH_TOKEN = os.getenv("CRASH_TOKEN") + QEMU_ARGS += " -qtest stdio -monitor none -serial none " + minimize_trace(sys.argv[-2], sys.argv[-1]) diff --git a/scripts/oss-fuzz/output_reproducer.py b/scripts/oss-fuzz/output_reproducer.py new file mode 100755 index 000000000..3608b0600 --- /dev/null +++ b/scripts/oss-fuzz/output_reproducer.py @@ -0,0 +1,160 @@ +#!/usr/bin/env python3 +# -*- coding: utf-8 -*- + +""" +Convert plain qtest traces to C or Bash reproducers + +Use this to help build bug-reports or create in-tree reproducers for bugs. +Note: This will not format C code for you. Pipe the output through +clang-format -style="{BasedOnStyle: llvm, IndentWidth: 4, ColumnLimit: 90}" +or similar +""" + +import sys +import os +import argparse +import textwrap +from datetime import date + +__author__ = "Alexander Bulekov <alxndr@bu.edu>" +__copyright__ = "Copyright (C) 2021, Red Hat, Inc." +__license__ = "GPL version 2 or (at your option) any later version" + +__maintainer__ = "Alexander Bulekov" +__email__ = "alxndr@bu.edu" + + +def c_header(owner): + return """/* + * Autogenerated Fuzzer Test Case + * + * Copyright (c) {date} {owner} + * + * This work is licensed under the terms of the GNU GPL, version 2 or later. + * See the COPYING file in the top-level directory. + */ + +#include "qemu/osdep.h" + +#include "libqos/libqtest.h" + + """.format(date=date.today().year, owner=owner) + +def c_comment(s): + """ Return a multi-line C comment. Assume the text is already wrapped """ + return "/*\n * " + "\n * ".join(s.splitlines()) + "\n*/" + +def print_c_function(s): + print("/* ") + for l in s.splitlines(): + print(" * {}".format(l)) + +def bash_reproducer(path, args, trace): + result = '\\\n'.join(textwrap.wrap("cat << EOF | {} {}".format(path, args), + 72, break_on_hyphens=False, + drop_whitespace=False)) + for l in trace.splitlines(): + result += "\n" + '\\\n'.join(textwrap.wrap(l,72,drop_whitespace=False)) + result += "\nEOF" + return result + +def c_reproducer(name, args, trace): + result = [] + result.append("""static void {}(void)\n{{""".format(name)) + + # libqtest will add its own qtest args, so get rid of them + args = args.replace("-accel qtest","") + args = args.replace(",accel=qtest","") + args = args.replace("-machine accel=qtest","") + args = args.replace("-qtest stdio","") + result.append("""QTestState *s = qtest_init("{}");""".format(args)) + for l in trace.splitlines(): + param = l.split() + cmd = param[0] + if cmd == "write": + buf = param[3][2:] #Get the 0x... buffer and trim the "0x" + assert len(buf)%2 == 0 + bufbytes = [buf[i:i+2] for i in range(0, len(buf), 2)] + bufstring = '\\x'+'\\x'.join(bufbytes) + addr = param[1] + size = param[2] + result.append("""qtest_bufwrite(s, {}, "{}", {});""".format( + addr, bufstring, size)) + elif cmd.startswith("in") or cmd.startswith("read"): + result.append("qtest_{}(s, {});".format( + cmd, param[1])) + elif cmd.startswith("out") or cmd.startswith("write"): + result.append("qtest_{}(s, {}, {});".format( + cmd, param[1], param[2])) + elif cmd == "clock_step": + if len(param) ==1: + result.append("qtest_clock_step_next(s);") + else: + result.append("qtest_clock_step(s, {});".format(param[1])) + result.append("qtest_quit(s);\n}") + return "\n".join(result) + +def c_main(name, arch): + return """int main(int argc, char **argv) +{{ + const char *arch = qtest_get_arch(); + + g_test_init(&argc, &argv, NULL); + + if (strcmp(arch, "{arch}") == 0) {{ + qtest_add_func("fuzz/{name}",{name}); + }} + + return g_test_run(); +}}""".format(name=name, arch=arch) + +def main(): + parser = argparse.ArgumentParser() + group = parser.add_mutually_exclusive_group() + group.add_argument("-bash", help="Only output a copy-pastable bash command", + action="store_true") + group.add_argument("-c", help="Only output a c function", + action="store_true") + parser.add_argument('-owner', help="If generating complete C source code, \ + this specifies the Copyright owner", + nargs='?', default="<name of author>") + parser.add_argument("-no_comment", help="Don't include a bash reproducer \ + as a comment in the C reproducers", + action="store_true") + parser.add_argument('-name', help="The name of the c function", + nargs='?', default="test_fuzz") + parser.add_argument('input_trace', help="input QTest command sequence \ + (stdin by default)", + nargs='?', type=argparse.FileType('r'), + default=sys.stdin) + args = parser.parse_args() + + qemu_path = os.getenv("QEMU_PATH") + qemu_args = os.getenv("QEMU_ARGS") + if not qemu_args or not qemu_path: + print("Please set QEMU_PATH and QEMU_ARGS environment variables") + sys.exit(1) + + bash_args = qemu_args + if " -qtest stdio" not in qemu_args: + bash_args += " -qtest stdio" + + arch = qemu_path.split("-")[-1] + trace = args.input_trace.read().strip() + + if args.bash : + print(bash_reproducer(qemu_path, bash_args, trace)) + else: + output = "" + if not args.c: + output += c_header(args.owner) + "\n" + if not args.no_comment: + output += c_comment(bash_reproducer(qemu_path, bash_args, trace)) + output += c_reproducer(args.name, qemu_args, trace) + if not args.c: + output += c_main(args.name, arch) + print(output) + + +if __name__ == '__main__': + main() diff --git a/scripts/oss-fuzz/reorder_fuzzer_qtest_trace.py b/scripts/oss-fuzz/reorder_fuzzer_qtest_trace.py new file mode 100755 index 000000000..b154a2550 --- /dev/null +++ b/scripts/oss-fuzz/reorder_fuzzer_qtest_trace.py @@ -0,0 +1,103 @@ +#!/usr/bin/env python3 +# -*- coding: utf-8 -*- + +""" +Use this to convert qtest log info from a generic fuzzer input into a qtest +trace that you can feed into a standard qemu-system process. Example usage: + +QEMU_FUZZ_ARGS="-machine q35,accel=qtest" QEMU_FUZZ_OBJECTS="*" \ + ./i386-softmmu/qemu-fuzz-i386 --fuzz-target=generic-pci-fuzz +# .. Finds some crash +QTEST_LOG=1 FUZZ_SERIALIZE_QTEST=1 \ +QEMU_FUZZ_ARGS="-machine q35,accel=qtest" QEMU_FUZZ_OBJECTS="*" \ + ./i386-softmmu/qemu-fuzz-i386 --fuzz-target=generic-pci-fuzz + /path/to/crash 2> qtest_log_output +scripts/oss-fuzz/reorder_fuzzer_qtest_trace.py qtest_log_output > qtest_trace +./i386-softmmu/qemu-fuzz-i386 -machine q35,accel=qtest \ + -qtest stdio < qtest_trace + +### Details ### + +Some fuzzer make use of hooks that allow us to populate some memory range, just +before a DMA read from that range. This means that the fuzzer can produce +activity that looks like: + [start] read from mmio addr + [end] read from mmio addr + [start] write to pio addr + [start] fill a DMA buffer just in time + [end] fill a DMA buffer just in time + [start] fill a DMA buffer just in time + [end] fill a DMA buffer just in time + [end] write to pio addr + [start] read from mmio addr + [end] read from mmio addr + +We annotate these "nested" DMA writes, so with QTEST_LOG=1 the QTest trace +might look something like: +[R +0.028431] readw 0x10000 +[R +0.028434] outl 0xc000 0xbeef # Triggers a DMA read from 0xbeef and 0xbf00 +[DMA][R +0.034639] write 0xbeef 0x2 0xAAAA +[DMA][R +0.034639] write 0xbf00 0x2 0xBBBB +[R +0.028431] readw 0xfc000 + +This script would reorder the above trace so it becomes: +readw 0x10000 +write 0xbeef 0x2 0xAAAA +write 0xbf00 0x2 0xBBBB +outl 0xc000 0xbeef +readw 0xfc000 + +I.e. by the time, 0xc000 tries to read from DMA, those DMA buffers have already +been set up, removing the need for the DMA hooks. We can simply provide this +reordered trace via -qtest stdio to reproduce the input + +Note: this won't work for traces where the device tries to read from the same +DMA region twice in between MMIO/PIO commands. E.g: + [R +0.028434] outl 0xc000 0xbeef + [DMA][R +0.034639] write 0xbeef 0x2 0xAAAA + [DMA][R +0.034639] write 0xbeef 0x2 0xBBBB + +The fuzzer will annotate suspected double-fetches with [DOUBLE-FETCH]. This +script looks for these tags and warns the users that the resulting trace might +not reproduce the bug. +""" + +import sys + +__author__ = "Alexander Bulekov <alxndr@bu.edu>" +__copyright__ = "Copyright (C) 2020, Red Hat, Inc." +__license__ = "GPL version 2 or (at your option) any later version" + +__maintainer__ = "Alexander Bulekov" +__email__ = "alxndr@bu.edu" + + +def usage(): + sys.exit("Usage: {} /path/to/qtest_log_output".format((sys.argv[0]))) + + +def main(filename): + with open(filename, "r") as f: + trace = f.readlines() + + # Leave only lines that look like logged qtest commands + trace[:] = [x.strip() for x in trace if "[R +" in x + or "[S +" in x and "CLOSED" not in x] + + for i in range(len(trace)): + if i+1 < len(trace): + if "[DMA]" in trace[i+1]: + if "[DOUBLE-FETCH]" in trace[i+1]: + sys.stderr.write("Warning: Likely double fetch on line" + "{}.\n There will likely be problems " + "reproducing behavior with the " + "resulting qtest trace\n\n".format(i+1)) + trace[i], trace[i+1] = trace[i+1], trace[i] + for line in trace: + print(line.split("]")[-1].strip()) + + +if __name__ == '__main__': + if len(sys.argv) == 1: + usage() + main(sys.argv[1]) |