Skip to content

Writing an Adapter

Tyson Smith edited this page Nov 28, 2023 · 18 revisions

Overview

To run a fuzzer via Grizzly an Adapter must be created. Depending on the fuzzer and the use case an Adapter may only be a few lines of Python. Only the generate() method must be implemented but typically setup() is implemented as well.

The function of the Adapter is to act as the interface between Grizzly and the fuzzer. This includes handling input data (if provided), output data and execution of the fuzzer. A TestCase must be populated that Grizzly will then use to send the data to the browser. This must be done by the Adapter developer by using the API provided by TestCase and in more advanced situations ServerMap.

NOTE: The API is subject to change without notice, although this will likely happen infrequently.

The most basic Adapter only needs to add a single file to the TestCase and it must point to TestCase.entry_point (the test case entry point). In other words if the Adapter attaches a file like this TestCase.add_from_file(file, file_name=TestCase.entry_point) Grizzly will handle the rest. Note that add_from_file() takes copy=<bool> as an argument which will either copy or move (default) the file. The TestCase implementation can be found here.

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

Basic Adapter

This example should work with Domato with very little modification but it is meant to act as an example so some assumptions are made. This is not meant to be used as is.

basic_adapter.py

from pathlib import Path
from shutil import rmtree
from subprocess import check_output
from tempfile import mkdtemp

from grizzly.adapter import Adapter

class BasicExampleAdapter(Adapter):
    def setup(self, input_path, server_map):
        self.enable_harness()
        # create directory to temporarily store generated content
        self.fuzz["working"] = Path(mkdtemp(prefix="fuzz_gen_"))
        # command to run the fuzzer (generate test data)
        self.fuzz["cmd"] = [
            input_path,  # binary to call
            "--no_of_files", "1",
            "--output_dir", str(self.fuzz["working"])
        ]
        # This is also a good place to make content
        # in other directories accessible via the web server.
        # For example assume that "/test/data/" contains:
        # - a.html
        # - b/c.js
        # - media/d.jpg
        # Using server_map.set_include("inc",  "/test/data/")
        # it can be made accessible via a test case as:
        # - /inc/a.html
        # - /inc/sub/include.js
        # - media/d.jpg
        # Any content served via an 'include' is automatically added
        # to the test case.

    def generate(self, testcase, _):
        # launch fuzzer to generate a single file
        check_output(self.fuzz["cmd"])
        # lookup the newly generated file on disk
        gen_file = next(self.fuzz["working"].iterdir())
        # add file to the test case as the landing page (entry point)
        testcase.add_from_file(
            gen_file, file_name=testcase.entry_point, required=True, copy=False
        )

    def shutdown(self):
        # remove temporary working directory if needed
        if self.fuzz["working"].is_dir():
            rmtree(self.fuzz["working"], 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 fuzzer 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 will exit when complete returning 0 on success.

example_adapter.py

from contextlib import contextmanager
from pathlib import Path
from shutil import rmtree
from subprocess import DEVNULL, Popen, TimeoutExpired
from tempfile import mkdtemp

from grizzly.adapter import Adapter, AdapterError

class ExampleError(AdapterError):
    pass

class ExampleAdapter(Adapter):
    def setup(self, input_path, _):
        """perform necessary setup here"""
        self.enable_harness()
        self.fuzz["generator"] = FuzzGenerator(input_path)

    def generate(self, testcase, _):
        with self.fuzz["generator"].fuzzed_file() as file:
            testcase.add_from_file(file, file_name=testcase.entry_point, required=True)

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


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

    def __init__(self, tool):
        self._pool = Path(mkdtemp(prefix="fuzz_gen_"))
        # build the command to call the external file generator
        self._cmd = [tool, "--count", str(self.QUEUE_SIZE), "--out", str(self._pool)]
        self._queue = None
        self._worker = None
        self._launch()

    def _launch(self):
        # launch external file generator / fuzzer
        self._worker = Popen(self._cmd, shell=False, stdout=DEVNULL, stderr=DEVNULL)

    def close(self):
        if self._worker is not None:
            if self._worker.poll() is None:
                self._worker.terminate()
            self._worker.wait()
        rmtree(self._pool, ignore_errors=True)

    @contextmanager
    def fuzzed_file(self):
        if self._worker is not None:
            try:
                self._worker.wait(timeout=self.GEN_TIMEOUT)
            except TimeoutExpired:
                self._worker.terminate()
                raise ExampleError("Generator timed out (%ds)" % (self.GEN_TIMEOUT,)) from None
            if self._worker.returncode != 0:
                raise ExampleError("Generator returned %d" % (self._worker.returncode,))
            self._worker = None
            self._queue = list(self._pool.iterdir())
            if not self._queue:
                raise ExampleError("Generator failed to generate test cases")
        elif not self._queue:
            raise ExampleError("Queue is empty and worker is not running")
        pending = self._queue.pop()
        yield pending
        pending.unlink(missing_ok=True)
        if not self._queue:
            self._launch()

Package and Install

In order to run an adapter it must be installed. The first step is to create a package for use with Setuptools, more information can be found here.

Here is a simple example for demo purposes:

setup.py

from setuptools import setup

setup(
    name='grizzly-adapter-example',
    version='0.0.1',
    install_requires=[
        'grizzly-framework',
    ],
    entry_points={
       "grizzly_adapters": ["example = example_adapter:ExampleAdapter"]
    },
)

The directory structure for this should look like:

adapter_dev/
    example_adapter.py
    setup.py

To install the adapter for use:

python -m pip install -e adapter_dev

It should now be available via Grizzly (listed in Installed Adapters):

$ python -m grizzly -h
usage: __main__.py ...

positional arguments:
  binary                Firefox binary to run
  adapter               Installed Adapters: example
...