Update examples of the custom mutator

- Merge `examples/python_mutators` into `examples/custom_mutators`
- Remove `examples/python_mutators`
- Update existing examples to demonstrate new APIs
This commit is contained in:
h1994st 2020-03-04 01:09:37 -05:00
parent 42ce48db39
commit 38e7dd2b9e
11 changed files with 313 additions and 156 deletions

View File

@ -231,7 +231,7 @@ checks or alter some of the more exotic semantics of the tool:
performed with the custom mutator.
This feature allows to configure custom mutators which can be very helpful,
e.g. fuzzing XML or other highly flexible structured input.
Please see [custom_mutator.md](custom_mutator.md).
Please see [custom_mutators.md](custom_mutators.md).
- AFL_FAST_CAL keeps the calibration stage about 2.5x faster (albeit less
precise), which can help when starting a session against a slow target.

View File

@ -1,4 +1,22 @@
# A simple example for AFL_CUSTOM_MUTATOR_LIBRARY
# Examples for the custom mutator
This is a simple example for the AFL_CUSTOM_MUTATOR_LIBRARY feature.
For more information see [docs/custom_mutator.md](../docs/custom_mutator.md)
These are example and helper files for the custom mutator feature.
See [docs/python_mutators.md](../docs/custom_mutators.md) for more information
Note that if you compile with python3.7 you must use python3 scripts, and if
you use pyton2.7 to compile python2 scripts!
example.c - this is a simple example written in C and should be compiled to a
shared library
example.py - this is the template you can use, the functions are there but they
are empty
simple-chunk-replace.py - this is a simple example where chunks are replaced
common.py - this can be used for common functions and helpers.
the examples do not use this though. But you can :)
wrapper_afl_min.py - mutation of XML documents, loads XmlMutatorMin.py
XmlMutatorMin.py - module for XML mutation

View File

@ -7,6 +7,7 @@ from copy import deepcopy
from lxml import etree as ET
import random, re, io
###########################
# The XmlMutatorMin class #
###########################
@ -40,19 +41,19 @@ class XmlMutatorMin:
self.tree = None
# High-level mutators (no database needed)
hl_mutators_delete = [ "del_node_and_children", "del_node_but_children", "del_attribute", "del_content" ] # Delete items
hl_mutators_fuzz = ["fuzz_attribute"] # Randomly change attribute values
hl_mutators_delete = ["del_node_and_children", "del_node_but_children", "del_attribute", "del_content"] # Delete items
hl_mutators_fuzz = ["fuzz_attribute"] # Randomly change attribute values
# Exposed mutators
self.hl_mutators_all = hl_mutators_fuzz + hl_mutators_delete
def __parse_xml (self, xml):
def __parse_xml(self, xml):
""" Parse an XML string. Basic wrapper around lxml.parse() """
try:
# Function parse() takes care of comments / DTD / processing instructions / ...
tree = ET.parse(io.BytesIO(xml))
tree = ET.parse(io.BytesIO(xml))
except ET.ParseError:
raise RuntimeError("XML isn't well-formed!")
except LookupError as e:
@ -61,34 +62,34 @@ class XmlMutatorMin:
# Return a document wrapper
return tree
def __exec_among (self, module, functions, min_times, max_times):
def __exec_among(self, module, functions, min_times, max_times):
""" Randomly execute $functions between $min and $max times """
for i in xrange (random.randint (min_times, max_times)):
for i in xrange(random.randint(min_times, max_times)):
# Function names are mangled because they are "private"
getattr (module, "_XmlMutatorMin__" + random.choice(functions)) ()
getattr(module, "_XmlMutatorMin__" + random.choice(functions))()
def __serialize_xml (self, tree):
def __serialize_xml(self, tree):
""" Serialize a XML document. Basic wrapper around lxml.tostring() """
return ET.tostring(tree, with_tail=False, xml_declaration=True, encoding=tree.docinfo.encoding)
def __ver (self, version):
def __ver(self, version):
""" Helper for displaying lxml version numbers """
return ".".join(map(str, version))
def reset (self):
def reset(self):
""" Reset the mutator """
self.tree = deepcopy(self.input_tree)
def init_from_string (self, input_string):
def init_from_string(self, input_string):
""" Initialize the mutator from a XML string """
# Get a pointer to the top-element
@ -97,15 +98,15 @@ class XmlMutatorMin:
# Get a working copy
self.tree = deepcopy(self.input_tree)
def save_to_string (self):
def save_to_string(self):
""" Return the current XML document as UTF-8 string """
# Return a text version of the tree
return self.__serialize_xml(self.tree)
def __pick_element (self, exclude_root_node = False):
def __pick_element(self, exclude_root_node=False):
""" Pick a random element from the current document """
# Get a list of all elements, but nodes like PI and comments
@ -119,7 +120,7 @@ class XmlMutatorMin:
# Pick a random element
try:
elem_id = random.randint (start, len(elems) - 1)
elem_id = random.randint(start, len(elems) - 1)
elem = elems[elem_id]
except ValueError:
# Should only occurs if "exclude_root_node = True"
@ -127,8 +128,8 @@ class XmlMutatorMin:
return (elem_id, elem)
def __fuzz_attribute (self):
def __fuzz_attribute(self):
""" Fuzz (part of) an attribute value """
# Select a node to modify
@ -144,19 +145,19 @@ class XmlMutatorMin:
return
# Pick a random attribute
rand_attrib_id = random.randint (0, len(attribs) - 1)
rand_attrib_id = random.randint(0, len(attribs) - 1)
rand_attrib = attribs[rand_attrib_id]
# We have the attribute to modify
# Get its value
attrib_value = rand_elem.get(rand_attrib);
attrib_value = rand_elem.get(rand_attrib)
# print("- Value: " + attrib_value)
# Should we work on the whole value?
func_call = "(?P<func>[a-zA-Z:\-]+)\((?P<args>.*?)\)"
p = re.compile(func_call)
l = p.findall(attrib_value)
if random.choice((True,False)) and l:
if random.choice((True, False)) and l:
# Randomly pick one the function calls
(func, args) = random.choice(l)
# Split by "," and randomly pick one of the arguments
@ -236,29 +237,29 @@ class XmlMutatorMin:
# Modify the attribute
rand_elem.set(rand_attrib, new_value.decode("utf-8"))
def __del_node_and_children (self):
def __del_node_and_children(self):
""" High-level minimizing mutator
Delete a random node and its children (i.e. delete a random tree) """
self.__del_node(True)
def __del_node_but_children (self):
def __del_node_but_children(self):
""" High-level minimizing mutator
Delete a random node but its children (i.e. link them to the parent of the deleted node) """
self.__del_node(False)
def __del_node (self, delete_children):
def __del_node(self, delete_children):
""" Called by the __del_node_* mutators """
# Select a node to modify (but the root one)
(rand_elem_id, rand_elem) = self.__pick_element (exclude_root_node = True)
(rand_elem_id, rand_elem) = self.__pick_element(exclude_root_node=True)
# If the document includes only a top-level element
# Then we can't pick a element (given that "exclude_root_node = True")
# Then we can't pick a element (given that "exclude_root_node = True")
# Is the document deep enough?
if rand_elem is None:
@ -275,12 +276,12 @@ class XmlMutatorMin:
# Link children of the random (soon to be deleted) node to its parent
for child in rand_elem:
rand_elem.getparent().append(child)
# Remove the node
rand_elem.getparent().remove(rand_elem)
def __del_content (self):
def __del_content(self):
""" High-level minimizing mutator
Delete the attributes and children of a random node """
@ -294,8 +295,8 @@ class XmlMutatorMin:
# Reset the node
rand_elem.clear()
def __del_attribute (self):
def __del_attribute(self):
""" High-level minimizing mutator
Delete a random attribute from a random node """
@ -312,7 +313,7 @@ class XmlMutatorMin:
return
# Pick a random attribute
rand_attrib_id = random.randint (0, len(attribs) - 1)
rand_attrib_id = random.randint(0, len(attribs) - 1)
rand_attrib = attribs[rand_attrib_id]
# Log something
@ -322,8 +323,8 @@ class XmlMutatorMin:
# Delete the attribute
rand_elem.attrib.pop(rand_attrib)
def mutate (self, min=1, max=5):
def mutate(self, min=1, max=5):
""" Execute some high-level mutators between $min and $max times, then some medium-level ones """
# High-level mutation

View File

@ -19,19 +19,22 @@ import random
import os
import re
def randel(l):
if not l:
return None
return l[random.randint(0,len(l)-1)]
return l[random.randint(0, len(l)-1)]
def randel_pop(l):
if not l:
return None
return l.pop(random.randint(0,len(l)-1))
return l.pop(random.randint(0, len(l)-1))
def write_exc_example(data, exc):
exc_name = re.sub(r'[^a-zA-Z0-9]', '_', repr(exc))
if not os.path.exists(exc_name):
with open(exc_name, 'w') as f:
f.write(data)
f.write(data)

View File

@ -0,0 +1,177 @@
/*
New Custom Mutator for AFL++
Written by Khaled Yakdan <yakdan@code-intelligence.de>
Andrea Fioraldi <andreafioraldi@gmail.com>
Shengtuo Hu <h1994st@gmail.com>
*/
#include <stdint.h>
#include <stdlib.h>
#include <string.h>
static const char *commands[] = {
"GET",
"PUT",
"DEL",
};
static size_t data_size = 100;
void afl_custom_init(unsigned int seed) {
srand(seed);
}
/**
* Perform custom mutations on a given input
*
* (Optional for now. Required in the future)
*
* @param[in] buf Input data to be mutated
* @param[in] buf_size Size of input data
* @param[in] add_buf Buffer containing the additional test case
* @param[in] add_buf_size Size of the additional test case
* @param[out] mutated_out Buffer to store the mutated input
* @param[in] max_size Maximum size of the mutated output. The mutation must not
* produce data larger than max_size.
* @return Size of the mutated output.
*/
size_t afl_custom_fuzz(uint8_t *buf, size_t buf_size,
uint8_t *add_buf,size_t add_buf_size, // add_buf can be NULL
uint8_t *mutated_out, size_t max_size) {
// Make sure that the packet size does not exceed the maximum size expected by
// the fuzzer
size_t mutated_size = data_size <= max_size ? data_size : max_size;
// Randomly select a command string to add as a header to the packet
memcpy(mutated_out, commands[rand() % 3], 3);
// Mutate the payload of the packet
for (int i = 3; i < mutated_size; i++) {
mutated_out[i] = (data[i] + rand() % 10) & 0xff;
}
return mutated_size;
}
/**
* A post-processing function to use right before AFL writes the test case to
* disk in order to execute the target.
*
* (Optional) If this functionality is not needed, simply don't define this
* function.
*
* @param[in] buf Buffer containing the test case to be executed
* @param[in] buf_size Size of the test case
* @param[out] out_buf Pointer to the buffer containing the test case after
* processing. External library should allocate memory for out_buf. AFL++
* will release the memory after saving the test case.
* @return Size of the output buffer after processing
*/
size_t afl_custom_pre_save(uint8_t *buf, size_t buf_size, uint8_t **out_buf) {
size_t out_buf_size;
out_buf_size = buf_size;
// External mutator should allocate memory for `out_buf`
*out_buf = malloc(out_buf_size);
memcpy(*out_buf, buf, out_buf_size);
return out_buf_size;
}
uint8_t *trim_buf;
size_t trim_buf_size
int trimmming_steps;
int cur_step;
/**
* This method is called at the start of each trimming operation and receives
* the initial buffer. It should return the amount of iteration steps possible
* on this input (e.g. if your input has n elements and you want to remove
* them one by one, return n, if you do a binary search, return log(n),
* and so on...).
*
* If your trimming algorithm doesn't allow you to determine the amount of
* (remaining) steps easily (esp. while running), then you can alternatively
* return 1 here and always return 0 in post_trim until you are finished and
* no steps remain. In that case, returning 1 in post_trim will end the
* trimming routine. The whole current index/max iterations stuff is only used
* to show progress.
*
* (Optional)
*
* @param buf Buffer containing the test case
* @param buf_size Size of the test case
* @return The amount of possible iteration steps to trim the input
*/
int afl_custom_init_trim(uint8_t *buf, size_t buf_size) {
// We simply trim once
trimmming_steps = 1;
cur_step = 0;
trim_buf = buf;
trim_buf_size = buf_size;
return trimmming_steps;
}
/**
* This method is called for each trimming operation. It doesn't have any
* arguments because we already have the initial buffer from init_trim and we
* can memorize the current state in global variables. This can also save
* reparsing steps for each iteration. It should return the trimmed input
* buffer, where the returned data must not exceed the initial input data in
* length. Returning anything that is larger than the original data (passed
* to init_trim) will result in a fatal abort of AFLFuzz.
*
* (Optional)
*
* @param[out] out_buf Pointer to the buffer containing the trimmed test case.
* External library should allocate memory for out_buf. AFL++ will release
* the memory after saving the test case.
* @param[out] out_buf_size Pointer to the size of the trimmed test case
*/
void afl_custom_trim(uint8_t **out_buf, size_t* out_buf_size) {
*out_buf_size = trim_buf_size - 1;
// External mutator should allocate memory for `out_buf`
*out_buf = malloc(*out_buf_size);
// Remove the last byte of the trimming input
memcpy(*out_buf, trim_buf, *out_buf_size);
}
/**
* This method is called after each trim operation to inform you if your
* trimming step was successful or not (in terms of coverage). If you receive
* a failure here, you should reset your input to the last known good state.
*
* (Optional)
*
* @param success Indicates if the last trim operation was successful.
* @return The next trim iteration index (from 0 to the maximum amount of
* steps returned in init_trim)
*/
int afl_custom_post_trim(int success) {
if (success) {
++cur_step;
return cur_step;
}
return trimmming_steps;
}

View File

@ -16,26 +16,31 @@ file, You can obtain one at http://mozilla.org/MPL/2.0/.
import random
def init(seed):
'''
Called once when AFLFuzz starts up. Used to seed our RNG.
@type seed: int
@param seed: A 32-bit random value
'''
random.seed(seed)
return 0
def fuzz(buf, add_buf):
def fuzz(buf, add_buf, max_size):
'''
Called per fuzzing iteration.
@type buf: bytearray
@param buf: The buffer that should be mutated.
@type add_buf: bytearray
@param add_buf: A second buffer that can be used as mutation source.
@type max_size: int
@param max_size: Maximum size of the mutated output. The mutation must not
produce data larger than max_size.
@rtype: bytearray
@return: A new bytearray containing the mutated data
'''
@ -50,54 +55,68 @@ def fuzz(buf, add_buf):
# def init_trim(buf):
# '''
# Called per trimming iteration.
#
#
# @type buf: bytearray
# @param buf: The buffer that should be trimmed.
#
#
# @rtype: int
# @return: The maximum number of trimming steps.
# '''
# global ...
#
#
# # Initialize global variables
#
#
# # Figure out how many trimming steps are possible.
# # If this is not possible for your trimming, you can
# # return 1 instead and always return 0 in post_trim
# # until you are done (then you return 1).
#
#
# return steps
#
#
# def trim():
# '''
# Called per trimming iteration.
#
#
# @rtype: bytearray
# @return: A new bytearray containing the trimmed data.
# '''
# global ...
#
#
# # Implement the actual trimming here
#
#
# return bytearray(...)
#
#
# def post_trim(success):
# '''
# Called after each trimming operation.
#
#
# @type success: bool
# @param success: Indicates if the last trim operation was successful.
#
#
# @rtype: int
# @return: The next trim index (0 to max number of steps) where max
# number of steps indicates the trimming is done.
# '''
# global ...
#
#
# if not success:
# # Restore last known successful input, determine next index
# else:
# # Just determine the next index, based on what was successfully
# # removed in the last step
#
#
# return next_index
#
# def pre_save(buf):
# '''
# Called just before the execution to write the test case in the format
# expected by the target
#
# @type buf: bytearray
# @param buf: The buffer containing the test case to be executed
#
# @rtype: bytearray
# @return: The buffer containing the test case after
# '''
# return buf
#

View File

@ -16,27 +16,32 @@ file, You can obtain one at http://mozilla.org/MPL/2.0/.
import random
def init(seed):
'''
Called once when AFLFuzz starts up. Used to seed our RNG.
@type seed: int
@param seed: A 32-bit random value
'''
# Seed our RNG
random.seed(seed)
return 0
def fuzz(buf, add_buf):
def fuzz(buf, add_buf, max_size):
'''
Called per fuzzing iteration.
@type buf: bytearray
@param buf: The buffer that should be mutated.
@type add_buf: bytearray
@param add_buf: A second buffer that can be used as mutation source.
@type max_size: int
@param max_size: Maximum size of the mutated output. The mutation must not
produce data larger than max_size.
@rtype: bytearray
@return: A new bytearray containing the mutated data
'''
@ -45,10 +50,10 @@ def fuzz(buf, add_buf):
# Take a random fragment length between 2 and 32 (or less if add_buf is shorter)
fragment_len = random.randint(1, min(len(add_buf), 32))
# Determine a random source index where to take the data chunk from
rand_src_idx = random.randint(0, len(add_buf) - fragment_len)
# Determine a random destination index where to put the data chunk
rand_dst_idx = random.randint(0, len(buf))

View File

@ -1,49 +0,0 @@
/*
Simple Custom Mutator for AFL
Written by Khaled Yakdan <yakdan@code-intelligence.de>
This a simple mutator that assumes that the generates messages starting with
one of the three strings GET, PUT, or DEL followed by a payload. The mutator
randomly selects a commend and mutates the payload of the seed provided as
input.
*/
#include <stdint.h>
#include <stdlib.h>
#include <string.h>
static const char *commands[] = {
"GET",
"PUT",
"DEL",
};
static size_t data_size = 100;
size_t afl_custom_mutator(uint8_t *data, size_t size, uint8_t *mutated_out,
size_t max_size, unsigned int seed) {
// Seed the PRNG
srand(seed);
// Make sure that the packet size does not exceed the maximum size expected by
// the fuzzer
size_t mutated_size = data_size <= max_size ? data_size : max_size;
// Randomly select a command string to add as a header to the packet
memcpy(mutated_out, commands[rand() % 3], 3);
// Mutate the payload of the packet
for (int i = 3; i < mutated_size; i++) {
mutated_out[i] = (data[i] + rand() % 10) & 0xff;
}
return mutated_size;
}

View File

@ -9,11 +9,11 @@ __seed__ = "RANDOM"
__log__ = False
__log_file__ = "wrapper.log"
# AFL functions
# AFL functions
def log(text):
"""
Logger
Logger
"""
global __seed__
@ -24,6 +24,7 @@ def log(text):
with open(__log_file__, "a") as logf:
logf.write("[%s] %s\n" % (__seed__, text))
def init(seed):
"""
Called once when AFL starts up. Seed is used to identify the AFL instance in log files
@ -37,17 +38,18 @@ def init(seed):
# Create a global mutation class
try:
__mutator__ = XmlMutatorMin(__seed__, verbose=__log__)
__mutator__ = XmlMutatorMin(__seed__, verbose=__log__)
log("init(): Mutator created")
except RuntimeError as e:
log("init(): Can't create mutator: %s" % e.message)
def fuzz(buf, add_buf):
def fuzz(buf, add_buf, max_size):
"""
Called for each fuzzing iteration.
Called for each fuzzing iteration.
"""
global __mutator__
global __mutator__
# Do we have a working mutator object?
if __mutator__ is None:
@ -62,7 +64,7 @@ def fuzz(buf, add_buf):
try:
buf_str = str(buf)
log("fuzz(): AFL buffer converted to a string")
except:
except Exception:
via_buffer = False
log("fuzz(): Can't convert AFL buffer to a string")
@ -71,28 +73,28 @@ def fuzz(buf, add_buf):
try:
__mutator__.init_from_string(buf_str)
log("fuzz(): Mutator successfully initialized with AFL buffer (%d bytes)" % len(buf_str))
except:
except Exception:
via_buffer = False
log("fuzz(): Can't initialize mutator with AFL buffer")
# If init from AFL buffer wasn't succesful
if not via_buffer:
log("fuzz(): Returning unmodified AFL buffer")
return buf
log("fuzz(): Returning unmodified AFL buffer")
return buf
# Sucessful initialization -> mutate
try:
__mutator__.mutate(max=5)
log("fuzz(): Input mutated")
except:
except Exception:
log("fuzz(): Can't mutate input => returning buf")
return buf
# Convert mutated data to a array of bytes
try:
data = bytearray(__mutator__.save_to_string())
log("fuzz(): Mutated data converted as bytes")
except:
except Exception:
log("fuzz(): Can't convert mutated data to bytes => returning buf")
return buf
@ -100,8 +102,8 @@ def fuzz(buf, add_buf):
log("fuzz(): Returning %d bytes" % len(data))
return data
# Main (for debug)
# Main (for debug)
if __name__ == '__main__':
__log__ = True
@ -114,4 +116,3 @@ if __name__ == '__main__':
in_2 = bytearray("<abc abc123='456' abcCBA='ppppppppppppppppppppppppppppp'/>")
out = fuzz(in_1, in_2)
print(out)

View File

@ -1,18 +0,0 @@
These are example and helper files for the AFL_PYTHON_MODULE feature.
See [docs/python_mutators.md](../docs/python_mutators.md) for more information
Note that if you compile with python3.7 you must use python3 scripts, and if
you use pyton2.7 to compile python2 scripts!
example.py - this is the template you can use, the functions are there
but they are empty
simple-chunk-replace.py - this is a simple example where chunks are replaced
common.py - this can be used for common functions and helpers.
the examples do not use this though. But you can :)
wrapper_afl_min.py - mutation of XML documents, loads XmlMutatorMin.py
XmlMutatorMin.py - module for XML mutation