Skip to content

Writing an Adapter

Tyson Smith edited this page Jul 11, 2019 · 3 revisions

Basic Adapter

To write an adapter see the Adapter template and for a working example see the no-op example adapter.

This example will work with Domato with very little modification but it is mainly meant to act as an example so some assumptions are made.

import os
import shutil
import subprocess
import tempfile
import time

from grizzly.common import Adapter, TestFile

class BasicExampleAdapter(Adapter):
    NAME = "basic"

    def setup(self, _):
        # create directory to temporarily store generated content
        self.fuzz["tmp"] = tempfile.mkdtemp(prefix="fuzz_gen_")
        # command used to call fuzzer to generate output
        self.fuzz["cmd"] = [
            os.environ["FUZZTOOL"],  # binary to call
            "--no_of_files", "1",
            "--output_dir", self.fuzz["tmp"]

    def generate(self, testcase, *_):
        # launch fuzzer to generate a single file
        # lookup the name of the newly generated file on disk
        gen_file = os.path.join(self.fuzz["tmp"], os.listdir(self.fuzz["tmp"])[0])
        # create a TestFile from the generated file
        test_file = TestFile.from_file(gen_file, testcase.landing_page)
        # remove generated file now that the data has been added to a test file
        # add test file to the testcase

    def shutdown(self):
        # remove temporary working directory if needed
        if os.path.isdir(self.fuzz["tmp"]):
            shutil.rmtree(self.fuzz["tmp"], ignore_errors=True)

Advanced Adapter

To write an Adapter for an existing fuzzer/test case generator here is a generic example that can be modified. We currently use variations of this Adapter to run internal fuzzers such as Avalanche and Dharma and external fuzzers such as Domato and Radamsa.

This example assumes that the binary FUZZTOOL_PATH takes two arguments --count number of test cases to generate and --out the path to dump test cases in. It also assumes when launched it will generate single files HTML test cases and exit when complete returning 0 on success.

from contextlib import contextmanager
import os
import shutil
import subprocess
import tempfile
import time

from grizzly.common import Adapter, AdapterError, TestFile

class ExampleError(AdapterError):

class ExampleAdapter(Adapter):
    NAME = "example"

    def setup(self, _):
        """ perform necessary setup here"""
        if "FUZZTOOL_PATH" not in os.environ:
            raise ExampleError("FUZZTOOL_PATH environment variable not set!")
        self.fuzz["generator"] = FuzzGenerator(os.environ["FUZZTOOL_PATH"])

    def generate(self, testcase, *_):
        with self.fuzz["generator"].testcase() as fname:
            testcase.add_file(TestFile.from_file(fname, testcase.landing_page))

    def shutdown(self):
        if "generator" in self.fuzz:

class FuzzGenerator(object):
    GEN_TIMEOUT = 120  # timeout for external test case generator
    QUEUE_SIZE = 10  # number of test cases to generate per batch

    def __init__(self, tool):
        self._pool = tempfile.mkdtemp(prefix="fuzz_gen_")
        # build the command to call the external test case generator
        # make sure to put tests in the correct temp path
        self._cmd = [tool, "--count", str(self.QUEUE_SIZE), "--out", self._pool]
        self._null = open(os.devnull, "w")
        self._queue = None
        self._worker = None

    def _launch(self):
        # launch external test case generator / fuzzer
        self._worker = subprocess.Popen(self._cmd, shell=False, stdout=self._null, stderr=self._null)

    def close(self):
        if self._worker is not None:
            if self._worker.poll() is None:
        if os.path.isdir(self._pool):
            shutil.rmtree(self._pool, ignore_errors=True)

    def testcase(self):
        assert self._queue or self._worker
        if self._worker is not None:
            deadline = time.time() + self.GEN_TIMEOUT
            while self._worker.poll() is None:
                if time.time() > deadline:
                    raise ExampleError("Timeout (%ds) waiting for generator" % (self.GEN_TIMEOUT,))
            if self._worker.returncode != 0:
                raise ExampleError("Test case generator returned %d" % (self._worker.returncode,))
            self._worker = None
            self._queue = os.listdir(self._pool)
            if not self._queue:
                raise ExampleError("Test case generator failed to generate test cases")
        pending = os.path.join(self._pool, self._queue.pop())
            yield pending
        if not self._queue:
Clone this wiki locally
You can’t perform that action at this time.