The layout submodule of the qcow2 package creates a random valid image, randomly selects some amount of its fields, fuzzes them and write the fuzzed image to the file. Fuzzing process can be controlled by an external configuration.
Now only a header, header extensions and a backing file name are generated. Signed-off-by: Maria Kustova <mari...@catit.be> --- tests/image-fuzzer/qcow2/layout.py | 359 +++++++++++++++++++++++++++++++++++++ 1 file changed, 359 insertions(+) create mode 100644 tests/image-fuzzer/qcow2/layout.py diff --git a/tests/image-fuzzer/qcow2/layout.py b/tests/image-fuzzer/qcow2/layout.py new file mode 100644 index 0000000..2bad223 --- /dev/null +++ b/tests/image-fuzzer/qcow2/layout.py @@ -0,0 +1,359 @@ +# Generator of fuzzed qcow2 images +# +# Copyright (C) 2014 Maria Kustova <mari...@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 random +import struct +import fuzz + +MAX_IMAGE_SIZE = 10*2**20 +# Standard sizes +UINT32_S = 4 +UINT64_S = 8 + + +class Field(object): + """Atomic image element (field) + + The class represents an image field as quadruple of a data format + of value necessary for its packing to binary form, an offset from + the beginning of the image, a value and a name. + + The field can be iterated as a list [format, offset, value] + """ + __slots__ = ('fmt', 'offset', 'value', 'name') + + def __init__(self, fmt, offset, val, name): + self.fmt = fmt + self.offset = offset + self.value = val + self.name = name + + def __iter__(self): + return (x for x in [self.fmt, self.offset, self.value]) + + def __repr__(self): + return "Field(fmt='%s', offset=%d, value=%s, name=%s)" % \ + (self.fmt, self.offset, str(self.value), self.name) + + +class FieldsList(object): + """List of fields + + The class allows access to a field in the list by its name and joins + several list in one via in-place addition + """ + def __init__(self, meta_data=None): + if meta_data is None: + self.data = [] + else: + self.data = [Field(f[0], f[1], f[2], f[3]) + for f in meta_data] + + def __getitem__(self, name): + return [x for x in self.data if x.name == name] + + def __iter__(self): + return (x for x in self.data) + + def __iadd__(self, other): + self.data += other.data + return self + + def __len__(self): + return len(self.data) + + +class Image(object): + """ Qcow2 image object + + This class allows to create qcow2 images with random valid structures and + values, fuzz them via external qcow2.fuzz module and write the result to + a file + """ + @staticmethod + def _size_params(): + """Generate a random image size aligned to a random correct + cluster size + """ + cluster_bits = random.randrange(9, 21) + cluster_size = 1 << cluster_bits + img_size = random.randrange(0, MAX_IMAGE_SIZE + 1, + cluster_size) + return [cluster_bits, img_size] + + @staticmethod + def _header(cluster_bits, img_size, backing_file_name=None): + """Generate a random valid header""" + meta_header = [ + ['>4s', 0, "QFI\xfb", 'magic'], + ['>I', 4, random.randint(2, 3), 'version'], + ['>Q', 8, 0, 'backing_file_offset'], + ['>I', 16, 0, 'backing_file_size'], + ['>I', 20, cluster_bits, 'cluster_bits'], + ['>Q', 24, img_size, 'size'], + ['>I', 32, 0, 'crypt_method'], + ['>I', 36, 0, 'l1_size'], + ['>Q', 40, 0, 'l1_table_offset'], + ['>Q', 48, 0, 'refcount_table_offset'], + ['>I', 56, 0, 'refcount_table_clusters'], + ['>I', 60, 0, 'nb_snapshots'], + ['>Q', 64, 0, 'snapshots_offset'], + ['>Q', 72, 0, 'incompatible_features'], + ['>Q', 80, 0, 'compatible_features'], + ['>Q', 88, 0, 'autoclear_features'], + # Only refcount_order = 4 is supported by current (07.2014) + # implementation of QEMU + ['>I', 96, 4, 'refcount_order'], + ['>I', 100, 0, 'header_length'] + ] + v_header = FieldsList(meta_header) + + if v_header['version'][0].value == 2: + v_header['header_length'][0].value = 72 + else: + v_header['incompatible_features'][0].value = random.getrandbits(2) + v_header['compatible_features'][0].value = random.getrandbits(1) + v_header['header_length'][0].value = 104 + + max_header_len = struct.calcsize(v_header['header_length'][0].fmt) + \ + v_header['header_length'][0].offset + end_of_extension_area_len = 2*UINT32_S + free_space = (1 << cluster_bits) - (max_header_len + + end_of_extension_area_len) + # If the backing file name specified and there is enough space for it + # in the first cluster, then it's placed in the very end of the first + # cluster. + if (backing_file_name is not None) and \ + (free_space >= len(backing_file_name)): + v_header['backing_file_size'][0].value = len(backing_file_name) + v_header['backing_file_offset'][0].value = (1 << cluster_bits) - \ + len(backing_file_name) + + return v_header + + @staticmethod + def _backing_file_name(header, backing_file_name=None): + """Add the name of the backing file at the offset specified + in the header + """ + if (backing_file_name is not None) and \ + (not header['backing_file_offset'][0].value == 0): + data_len = len(backing_file_name) + data_fmt = '>' + str(data_len) + 's' + data_field = FieldsList([ + [data_fmt, header['backing_file_offset'][0].value, + backing_file_name, 'bf_name'] + ]) + else: + data_field = FieldsList() + + return data_field + + @staticmethod + def _backing_file_format(header, backing_file_fmt=None): + """Generate the header extension for the backing file + format + """ + # Calculation of the free space available in the first cluster + offset = struct.calcsize(header['header_length'][0].fmt) + \ + header['header_length'][0].offset + end_of_extension_area_len = 2*UINT32_S + high_border = (header['backing_file_offset'][0].value or + ((1 << header['cluster_bits'][0].value) - 1)) - \ + end_of_extension_area_len + free_space = high_border - offset + ext_size = 2*UINT32_S + ((len(backing_file_fmt) + 7) & ~7) + + if (backing_file_fmt is not None) and (free_space >= ext_size): + ext_data_len = len(backing_file_fmt) + ext_data_fmt = '>' + str(ext_data_len) + 's' + ext_padding_len = 7 - (ext_data_len - 1) % 8 + ext = FieldsList([ + ['>I', offset, 0xE2792ACA, 'ext_magic'], + ['>I', offset + UINT32_S, ext_data_len, 'ext_length'], + [ext_data_fmt, offset + UINT32_S*2, backing_file_fmt, + 'bf_format'] + ]) + offset = ext['bf_format'][0].offset + \ + struct.calcsize(ext['bf_format'][0].fmt) + ext_padding_len + else: + ext = FieldsList() + return (ext, offset) + + @staticmethod + def _feature_name_table(header, offset): + """Generate a random header extension for names of features used in + the image + """ + def gen_feat_ids(): + """Return random feature type and feature bit""" + return (random.randint(0, 2), random.randint(0, 63)) + + end_of_extension_area_len = 2*UINT32_S + high_border = (header['backing_file_offset'][0].value or + (1 << header['cluster_bits'][0].value) - 1) - \ + end_of_extension_area_len + free_space = high_border - offset + # Sum of sizes of 'magic' and 'length' header extension fields + ext_header_len = 2*UINT32_S + fnt_entry_size = 6*UINT64_S + num_fnt_entries = min(10, (free_space - ext_header_len)/fnt_entry_size) + if not num_fnt_entries == 0: + feature_tables = [] + feature_ids = [] + inner_offset = offset + ext_header_len + feat_name = 'some cool feature' + while len(feature_tables) < num_fnt_entries*3: + feat_type, feat_bit = gen_feat_ids() + # Remove duplicates + while (feat_type, feat_bit) in feature_ids: + feat_type, feat_bit = gen_feat_ids() + feature_ids.append((feat_type, feat_bit)) + feat_fmt = '>' + str(len(feat_name)) + 's' + feature_tables += [['B', inner_offset, + feat_type, 'feature_type'], + ['B', inner_offset + 1, feat_bit, + 'feature_bit_number'], + [feat_fmt, inner_offset + 2, + feat_name, 'feature_name'] + ] + inner_offset += fnt_entry_size + # No padding for the extension is necessary, because + # the extension length is multiple of 8 + ext = FieldsList([ + ['>I', offset, 0x6803f857, 'ext_magic'], + ['>I', offset + UINT32_S, len(feature_tables)*48, 'ext_length'] + ] + feature_tables) + offset = inner_offset + else: + ext = FieldsList() + + return (ext, offset) + + @staticmethod + def _end_of_extension_area(offset): + """Generate a mandatory header extension marking end of header + extensions + """ + ext = FieldsList([ + ['>I', offset, 0, 'ext_magic'], + ['>I', offset + UINT32_S, 0, 'ext_length'] + ]) + return ext + + def __init__(self, backing_file_name=None, backing_file_fmt=None): + """Create a random valid qcow2 image with the correct inner structure + and allowable values + """ + # Image size is saved as an attribute for the runner needs + cluster_bits, self.image_size = self._size_params() + # Saved as an attribute, because it's necessary for writing + self.cluster_size = 1 << cluster_bits + self.header = self._header(cluster_bits, self.image_size, + backing_file_name) + self.backing_file_name = self._backing_file_name(self.header, + backing_file_name) + self.backing_file_format, \ + offset = self._backing_file_format(self.header, + backing_file_fmt) + self.feature_name_table, \ + offset = self._feature_name_table(self.header, offset) + self.end_of_extension_area = self._end_of_extension_area(offset) + # Container for entire image + self.data = FieldsList() + # Percentage of fields will be fuzzed + self.bias = random.uniform(0.2, 0.5) + + def __iter__(self): + return (x for x in [self.header, + self.backing_file_format, + self.feature_name_table, + self.end_of_extension_area, + self.backing_file_name]) + + def _join(self): + """Join all image structure elements as header, tables, etc in one + list of fields + """ + if len(self.data) == 0: + for v in self: + self.data += v + + def fuzz(self, fields_to_fuzz=None): + """Fuzz an image by corrupting values of a random subset of its fields + + Without parameters the method fuzzes an entire image. + If 'fields_to_fuzz' is specified then only fields in this list will be + fuzzed. 'fields_to_fuzz' can contain both individual fields and more + general image elements as a header or tables. + In the first case the field will be fuzzed always. + In the second a random subset of fields will be selected and fuzzed. + """ + def coin(): + """Return boolean value proportional to a portion of fields to be + fuzzed + """ + return random.random() < self.bias + + if fields_to_fuzz is None: + self._join() + for field in self.data: + if coin(): + field.value = getattr(fuzz, field.name)(field.value) + else: + for item in fields_to_fuzz: + if len(item) == 1: + for field in self.__dict__[item[0]]: + if coin(): + field.value = getattr(fuzz, + field.name)(field.value) + else: + for field in self.__dict__[item[0]][item[1]]: + try: + field.value = getattr(fuzz, field.name)( + field.value) + except AttributeError: + # Some fields can be skipped depending on + # references, e.g. FNT header extension is not + # generated for a feature mask header field + # equal to zero + pass + + def write(self, filename): + """Writes an entire image to the file""" + image_file = open(filename, 'w') + self._join() + for field in self.data: + image_file.seek(field.offset) + image_file.write(struct.pack(field.fmt, field.value)) + image_file.seek(0, 2) + size = image_file.tell() + roundup = (size + self.cluster_size - 1) & ~(self.cluster_size - 1) + if roundup > size: + image_file.seek(roundup - 1) + image_file.write("\0") + image_file.close() + + +def create_image(test_img_path, backing_file_name=None, backing_file_fmt=None, + fields_to_fuzz=None): + """Create a fuzzed image and write it to the specified file""" + image = Image(backing_file_name, backing_file_fmt) + image.fuzz(fields_to_fuzz) + image.write(test_img_path) + return image.image_size -- 1.9.3
