1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
|
#!/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
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" {} 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=
""".format((sys.argv[0])))
def check_if_trace_crashes(trace, path):
global CRASH_TOKEN
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)
stdo = rc.communicate()[0]
output = stdo.decode('unicode_escape')
if rc.returncode == 137: # Timed Out
return False
if len(output.splitlines()) < 2:
return False
if CRASH_TOKEN is None:
CRASH_TOKEN = output.splitlines()[-2]
return CRASH_TOKEN in output
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))
print("Identifying Crashes by this string: {}".format(CRASH_TOKEN))
i = 0
newtrace = trace[:]
# For each line
while i < len(newtrace):
# 1.) Try to remove it completely and reproduce the crash. If it works,
# we're done.
prior = newtrace[i]
print("Trying to remove {}".format(newtrace[i]))
# Try to remove the line completely
newtrace[i] = ""
if check_if_trace_crashes(newtrace, outpath):
i += 1
continue
newtrace[i] = prior
# 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
# 3.) If it is a qtest write command: write addr len data, try to split
# it into two separate write commands. If splitting the write down the
# middle does not work, try to move the pivot "left" and retry, until
# there is no space left. The idea is to prune unneccessary bytes from
# long writes, while accommodating arbitrary MemoryRegion access sizes
# and alignments.
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:
leftlength = int(length/2)
rightlength = length - leftlength
newtrace.insert(i+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
else:
leftlength -= 1
rightlength += 1
if check_if_trace_crashes(newtrace, outpath):
i -= 1
else:
newtrace[i] = prior
del newtrace[i+1]
i += 1
check_if_trace_crashes(newtrace, outpath)
if __name__ == '__main__':
if len(sys.argv) < 3:
usage()
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[1], sys.argv[2])
|
