Fuzzing Firm (python3 fuzzing.py fuzz_template.json)

fuzz_template.json

{
  "PROJECT": "afl",
  "ROOT": ".",
  "FUZZER": "$ROOT/afl/afl-fuzz",
  "SCHEDULE": "",
  "PREPROCESS": "",
  "BINARY": "imginfo",
  "BINVERSION": "19",
  "BINOPT": "-f @@",
  "BINPATH": "$ROOT/dataset/apps/$BINARY",
  "SEEDNAME":"empty",
  "SEEDDIR":"$ROOT/seed/$SEEDNAME",
  "MACHINE":"diavola",
  "TIMELIMIT":"24h",
  "ID":"0",
  "EXPNUM": "4",
  "TIMEOUT": "10000",
  "MEMLIMIT": "1024",
  "PARAM": "-i $SEEDDIR -t $TIMEOUT -m $MEMLIMIT -l $TIMELIMIT -o @LABEL -- $BINPATH $BINOPT",
  "OUTPUT": "$ROOT/results/$PROJECT"
}

fuzzing.py

import json
import subprocess, os
import resource
import logging as log

class Config:    
    def __init__(self, path):
        with open(path, "r") as fp:
            self.rawdata = json.load(fp)
        self.variables = self.__make_variables(self.rawdata)
        self.config = self.__make_config(self.rawdata, self.variables)
        log.debug(self.config)
        
    def __getitem__(self, item):
        return self.config[item]
    
    def __make_variables(self, rawdata):
        __variables = {}
        for k, v in rawdata.items():
            if type(v) == str:
                __variables["$" + k] = v
        return __variables
        
    def __make_config(self, rawdata, variables):
        __config = {}
        for k, v in self.rawdata.items():
            if type(v) == str:            
                __config[k] = self.__replace_param(v, variables)
            else:
                __config[k] = v
        while self.__need_parse(__config) == True:
            for k, v in __config.items():
                if type(v) == str:            
                    __config[k] = self.__replace_param(v, variables)
                else:
                    __config[k] = v                
        return __config
    
    def __need_parse(self, _config):
        print(_config)
        for v in _config.values():
            if "$" in v:
                return True
        return False
        
    def __replace_param(self, text, params):
        for i, j in params.items():
            text = text.replace(i, j)
        return text
        
    def has_scheduler(self):
        return "EXPNUM" in self.config.keys()
        
class Environment:
    def __init__(self, config):
        self.env = config["ENVIRONMENT"]
        self.machine = config["MACHINE"]
                
class Fuzzer:
    def __init__(self, config):
        self.process = config["FUZZER"]
        self.schedule = config["SCHEDULE"]
        self.binary = config["BINARY"]
        self.binopt = config["BINOPT"]        
        self.binpath = config["BINPATH"]
        self.seed = config["SEEDNAME"]
        self.seeddir = config["SEEDDIR"]
        self.timelimit = config["TIMELIMIT"]
        self.timeout = config["TIMEOUT"]
        self.memlimit = config["MEMLIMIT"]
                
    def run(self, param, environment):
        command = "{} {}".format(self.process, param)
        log.info(command)
        log.info(environment.env)
        resource.setrlimit(resource.RLIMIT_CORE, (-1, -1)) # For core dump
        print(command)
        subprocess.Popen(command, env=environment.env, shell=True)
        
class Experiment:
    def __init__(self, id, config):
        self.project = config["PROJECT"]
        self.rootdir = os.path.abspath(config["ROOT"])
        self.preproc = os.path.abspath(config["PREPROCESS"])
        self.binver = config["BINVERSION"]
        self.id = id
        self.output = os.path.abspath(config["OUTPUT"])
        self.fuzzer = Fuzzer(config)
        self.env = Environment(config)        
        _output = self.output + "/" + self.__get_label(self.id)
        self.param = config["PARAM"].replace("@LABEL", _output)
                
        try:                                                                                    
            os.mkdir(self.output)
        except OSError:                                                                         
            pass  
                
    def __get_label(self, id):
        return "%s_%s_%s_%s_%02d" % ("{}-{}".format(self.fuzzer.binary, self.binver), 
                        self.fuzzer.seed, self.env.machine, self.fuzzer.timelimit, id)
                        
    def preprocess(self):
        if self.preproc != "":
            os.system(self.preproc)
        
    def run(self):
        self.preprocess()        
        self.fuzzer.run(self.param, self.env)

class Scheduler:
    def __init__(self):
        pass
        
    def run(self, config_path):
        config = Config(config_path) # ChooseConfig
        if config.has_scheduler():
            exps = int(config["EXPNUM"])
            start = int(config["ID"])
            for e in list(range(exps)):
                experiment = Experiment(start + e, config) # Generate Experiment
                experiment.run()
        else:
            experiment = Experiment(int(config["ID"]), config) # Generate Experiment
            experiment.run()
            
        
def banner():
    print("    ______             _              ______                   ")
    print("|  ___|           (_)             |  ___|                  ")
    print("| |_ _   _ _________ _ __   __ _  | |_ __ _ _ __ _ __ ___  ")
    print("|  _| | | |_  /_  / | '_ \ / _` | |  _/ _` | '__| '_ ` _ \ ")
    print("| | | |_| |/ / / /| | | | | (_| | | || (_| | |  | | | | | |")
    print("\_|  \__,_/___/___|_|_| |_|\__, | \_| \__,_|_|  |_| |_| |_|")
    print("                            __/ |                          ")
    print("                           |___/                           ")
    
def usage():
    print("./fuzzing.py <config.json>")
    
if __name__ == "__main__":
    if len(os.sys.argv) < 2:
        usage()
        exit(1)
    banner()
    #log.basicConfig(level=log.DEBUG)
    sched = Scheduler()
    sched.run(os.sys.argv[1])