For my work on learning symbolic inputs, I wrote a variant of delta debugging which reduces parse trees based on an existing grammar. It works quite well and reasonably fast so far. I do not know whether it is relevant for the Debugging Book, and the code is quite lengthy and not didactically optimized, but thought I'd share it here anyway.
After the below code implementing the reduction, I'll also post an example processing CSV.
import itertools
import logging
import re
from fuzzingbook.GrammarFuzzer import tree_to_string
from fuzzingbook.Grammars import RE_NONTERMINAL, is_nonterminal
from fuzzingbook.Parser import EarleyParser
class Reducer:
def __init__(self, grammar,
concrete_input,
test,
expected_exc_type=Exception,
max_tries_for_generalization=50,
max_tries_for_validation=200,
max_retries_after_failed_validation=3):
self.grammar = grammar
self.test = test
self.max_tries_for_generalization = max_tries_for_generalization
self.max_tries_for_validation = max_tries_for_validation
self.max_retries_after_failed_validation = max_retries_after_failed_validation
self.expected_exc_type = expected_exc_type
self.logger = logging.getLogger("Reducer")
self.concrete_input = concrete_input
if max_tries_for_validation < max_tries_for_generalization:
self.max_tries_for_validation = 2 * max_tries_for_generalization
def reduce(self, concrete_input=None):
"""Implements the idea of Delta Debugging for input trees."""
self.logger.info("PHASE START: Reduction à la Delta Debugging.")
if concrete_input is None:
concrete_input = self.concrete_input
parse_tree = list(EarleyParser(self.grammar).parse(concrete_input))[0]
did_generalize = True
while did_generalize:
paths = [[0]]
did_generalize = False
while paths:
path = paths.pop()
node, children = Reducer.get_subtree(parse_tree, path)
tree_for_log = tree_to_string((node, children))
compatible_subtree_paths = []
for i, child in enumerate(children):
compatible_paths = self.find_paths(child, lambda p, t: t[0] == node)
if compatible_paths:
compatible_subtree_paths.extend(list(map(lambda p: path + [i] + p, compatible_paths)))
did_replace = False
# Step 1: Try to replace this subtree by one of its subtree
for compatible_path in compatible_subtree_paths:
subtree = Reducer.get_subtree(parse_tree, compatible_path)
maybe_new_input = Reducer.replace_subtree(parse_tree, subtree, path)
if self.test_input(tree_to_string(maybe_new_input)):
self.logger.info(f"Generalizing path {path}, "
f"replacing {tree_to_string((node, children))} "
f"with {tree_to_string(subtree)}")
parse_tree = maybe_new_input
self.logger.info(f"Current input: {tree_to_string(parse_tree)}")
did_replace, did_generalize = True, True
break
# Step 2: Try to choose another expansion in the grammar to replace the
# children set by just one child or fewer children.
if not did_replace:
expansions = [list(filter(lambda l: l, re.split(RE_NONTERMINAL, e)))
for e in self.grammar.get(node, [])]
current_expansion = [node for node in [c[0] for c in children]]
for elements in [e for e in expansions if e != current_expansion]:
# Find matching subtrees for each nonterminal
nonterminals = [e for e in elements if is_nonterminal(e)]
compatible_subtree_paths = [None] * len(nonterminals)
for k, nonterminal in enumerate(nonterminals):
compatible_subtree_paths[k] = []
for i, child in enumerate(children):
compatible_paths = self.find_paths(child, lambda p, t: t[0] == nonterminal)
if compatible_paths:
compatible_subtree_paths[k].extend(
list(map(lambda p: path + [i] + p, compatible_paths)))
# What itertools.product does:
# [[1, 2], [3, 4, 5]] -> [[1, 3], [1, 4], [1, 5], [2, 3], ...]
# Try all instantiations
for combination in itertools.product(*compatible_subtree_paths):
i = 0
new_children = []
for element in elements:
if not is_nonterminal(element):
if type(element) is not tuple:
element = (element, [])
new_children.append(element)
else:
new_children.append(Reducer.get_subtree(parse_tree, combination[i]))
i += 1
subtree = (node, new_children)
if Reducer.count_nodes((node, children)) <= Reducer.count_nodes((node, new_children)):
continue
maybe_new_input = Reducer.replace_subtree(parse_tree, subtree, path)
if self.test_input(tree_to_string(maybe_new_input)):
self.logger.info(f"Generalizing path {path}, "
f"replacing {tree_to_string((node, children))} "
f"with {tree_to_string(subtree)}")
parse_tree = maybe_new_input
self.logger.info(f"Current input: {str(parse_tree)}")
did_replace, did_generalize = True, True
break
if not did_replace:
self.logger.info(f"Cannot generalize path {path}, expression {tree_for_log}")
if children is not None:
for i, child in enumerate(children):
child_symbol, _ = child
if child_symbol in self.grammar:
paths.append(path + [i])
self.logger.info("PHASE END: Reduction à la Delta Debugging.")
self.concrete_input = tree_to_string(parse_tree)
return self.concrete_input
def test_input(self, concrete_input):
self.logger.debug(f"Testing {repr(concrete_input)}...")
try:
self.test(concrete_input)
except Exception as exc:
if issubclass(type(exc), self.expected_exc_type):
self.logger.debug(f"FAIL ({type(exc).__name__})")
return True
else:
self.logger.info(f"UNEXPECTED FAIL ({type(exc).__name__})")
return False
else:
self.logger.debug(f"PASS")
return False
def find_paths(self, tree, predicate, path=None):
"""
Return a list of all paths for which `predicate` holds.
`predicate` is a function `predicate`(`path`, `tree`), where
`path` denotes a subtree in `tree`. If `predicate()` returns
True, `path` is included in the returned list.
Taken from the Debugging Book.
"""
if path is None:
path = []
symbol, children = Reducer.get_subtree(tree, path)
if predicate(path, (symbol, children)):
return [path]
paths = []
if children is not None:
for i, child in enumerate(children):
child_symbol, _ = child
if child_symbol in self.grammar:
paths += self.find_paths(tree, predicate, path + [i])
return paths
@staticmethod
def get_subtree(tree, path):
"""Access a subtree based on `path` (a list of children numbers)"""
node, children = tree
if not path:
return tree
return Reducer.get_subtree(children[path[0]], path[1:])
@staticmethod
def replace_subtree(tree, replacement_tree, path):
"""Returns a symbolic input with a new tree where replacement_tree has been inserted at `path`"""
def recurse(tree, path):
node, children = tree
if not path:
return replacement_tree
head = path[0]
new_children = (children[:head] +
[recurse(children[head], path[1:])] +
children[head + 1:])
return node, new_children
return recurse(tree, path)
@staticmethod
def dfs(tree, action=print):
node, children = tree
action(tree)
for child in children:
Reducer.dfs(child, action)
@staticmethod
def count_nodes(parse_tree):
num = 0
def count(t):
nonlocal num
num += 1
Reducer.dfs(parse_tree, count)
return num
CSV_EBNF_GRAMMAR = {
"<start>": ["<csv-file>"],
"<csv-file>": ["<csv-record>*"],
"<csv-record>": ["<csv-string-list>\n"],
"<csv-string-list>": ["<raw-string>", "<raw-string>;<csv-string-list>"],
"<raw-string>": ["<spaces>", "<spaces>?<raw-field><spaces>?"],
"<raw-field>": ["<simple-field>", "<quoted-field>"],
"<simple-field>": ["<simple-character>*"],
"<simple-character>": [c for c in srange(string.printable) if c not in ["\n", ";", '"', " ", "\t", "\r"]],
"<quoted-field>": ['"<escaped-field>"'],
"<escaped-field>": ["<escaped-character>*"],
"<escaped-character>": [c for c in srange(string.printable) if c not in ['"']],
"<spaces>": [" ", " <spaces>"],
}
CSV_GRAMMAR = convert_ebnf_grammar(CSV_EBNF_GRAMMAR)
CSV_PARSER = EarleyParser(CSV_GRAMMAR)
class CSVFile:
def __init__(self, header: list, lines: list[list]):
self.header = header
self.lines = []
for line in lines:
the_dict = {}
for i, entry in enumerate(line):
the_dict[header[i]] = entry
self.lines.append(the_dict)
def __getitem__(self, item):
if type(item) is str:
return [line[item] for line in self.lines]
else:
return self.lines[item]
def __iter__(self):
return self.lines.__iter__()
def __str__(self):
result = " ; ".join(self.header)
if self.lines:
result += "\n"
result += "\n".join([" ; ".join(line.values()) for line in self.lines])
return result
def read_csv_file(inp):
tree = list(CSV_PARSER.parse(inp))[0]
fields = []
records = []
def collect_records(t):
nonlocal fields, records
node, children = t
if node == "<csv-record>":
record = []
for child in children:
fields = []
Reducer.dfs(child, collect_fields)
if fields:
record.append(fields)
records.extend(record)
def collect_fields(t):
nonlocal fields
node, children = t
if node == "<csv-string-list>":
fields.append(tree_to_string(children[0]))
Reducer.dfs(tree, collect_records)
if len(records) < 1:
raise SyntaxError("CSV file must contain at least a header line!")
return CSVFile(records[0], records[1:])