ファジングの基本的な仕組みを学ぶために、オープンソースであるboofuzzのコード読みをします。

テンプレートを見る

どのようなファジングを行うかユーザが定義するためのPython スクリプトを、こちらのサイトに倣ってテンプレート (fuzzer template) と呼ぶことにします。

まずは、コードを読む前にテンプレートを見てみます。リポジトリのexamplesディレクトリに複数のテンプレートが用意されているので、ここではhttp_simple.pyを見ます。

def main():
    session = Session(
        target=Target(connection=TCPSocketConnection("127.0.0.1", 80)),
    )

    define_proto(session=session)

    session.fuzz()

def define_proto(session):
    # disable Black formatting to keep custom indentation
    # fmt: off
    req = Request("HTTP-Request", children=(
        Block("Request-Line", children=(
            Group(name="Method", values=["GET", "HEAD", "POST", "PUT", "DELETE", "CONNECT", "OPTIONS", "TRACE"]),
            Delim(name="space-1", default_value=" "),
            String(name="URI", default_value="/index.html"),
            Delim(name="space-2", default_value=" "),
            String(name="HTTP-Version", default_value="HTTP/1.1"),
            Static(name="CRLF", default_value="\r\n"),
        )),
        Block("Host-Line", children=(
            String(name="Host-Key", default_value="Host:"),
            Delim(name="space", default_value=" "),
            String(name="Host-Value", default_value="example.com"),
            Static(name="CRLF", default_value="\r\n"),
        )),
        Static(name="CRLF", default_value="\r\n"),
    ))
    # fmt: on

    session.connect(req)

（中略）

if __name__ == "__main__":
    main()

上記のテンプレートは、見る限りでは、

localhostの80番ポートをターゲットに設定しつつ、Sessionクラスのインスタンスsessionを定義する
define_proto()関数によって、HTTPプロトコルでファジングを行うよう設定する
session.fuzz()でファジングを実行する

以上の処理を記述していると思われます。

fuzz()のコード読み

実際にファジングを行う処理はsession.fuzz()関数にて行われてそうなので、boofuzz/boofuzz/sessions.pyにて定義されているfuzz()関数からコードを読んでみます。

    def fuzz(self, name=None, max_depth=None):
        if name is None or name == "":
            self._main_fuzz_loop(self._generate_mutations_indefinitely(max_depth=max_depth))
        else:
            self.fuzz_by_name(name=name)

引数nameは、http_simple.pyでは特に設定されていないので、_main_fuzz_loop()関数に着目します。

_main_fuzz_loop()の引数_generate_mutations_indefinitely()は、イテレータを返すジェネレータ関数です。

また、_main_fuzz_loop()は名前からファジングに関する何らかの処理を繰り返す関数であると推測できます。

_main_fuzz_loop()

とりあえず、_main_fuzz_loop()関数を読み進めます。

    def _main_fuzz_loop(self, fuzz_case_iterator):
        """Execute main fuzz logic; takes an iterator of test cases.
        Preconditions: `self.total_mutant_index` and `self.total_num_mutations` are set properly.
        Args:
            fuzz_case_iterator (Iterable): An iterator that walks through fuzz cases and yields MutationContext objects.
                 See _iterate_single_node() for details.
        Returns:
            None
        """
        self.server_init()

        try:
            self._start_target(self.targets[0])

            if self._reuse_target_connection:
                self.targets[0].open()
            self.num_cases_actually_fuzzed = 0
            self.start_time = time.time()
            for mutation_context in fuzz_case_iterator:
                if self.total_mutant_index < self._index_start:
                    continue

                # Check restart interval
                if (
                    self.num_cases_actually_fuzzed
                    and self.restart_interval
                    and self.num_cases_actually_fuzzed % self.restart_interval == 0
                ):
                    self._fuzz_data_logger.open_test_step("restart interval of %d reached" % self.restart_interval)
                    self._restart_target(self.targets[0])

                self._fuzz_current_case(mutation_context)

                self.num_cases_actually_fuzzed += 1

                if self._index_end is not None and self.total_mutant_index >= self._index_end:
                    break

            if self._reuse_target_connection:
                self.targets[0].close()

            if self._keep_web_open and self.web_port is not None:
                self.end_time = time.time()
                print(
                    "\nFuzzing session completed. Keeping webinterface up on {}:{}".format(
                        self.web_address, self.web_port
                    ),
                    "\nPress ENTER to close webinterface",
                )
                input()
        except KeyboardInterrupt:
            # TODO: should wait for the end of the ongoing test case, and stop gracefully netmon and procmon
            self.export_file()
            self._fuzz_data_logger.log_error("SIGINT received ... exiting")
            raise
        except exception.BoofuzzRestartFailedError:
            self._fuzz_data_logger.log_error("Restarting the target failed, exiting.")
            self.export_file()
            raise
        except exception.BoofuzzTargetConnectionFailedError:
            # exception should have already been handled but rethrown in order to escape test run
            pass
        except Exception:
            self._fuzz_data_logger.log_error("Unexpected exception! {0}".format(traceback.format_exc()))
            self.export_file()
            raise
        finally:
            self._fuzz_data_logger.close_test()

この関数のコメントによると、引数fuzz_case_iteratorはファズケースを走査し、MutationContextオブジェクトを生成するイテレータとのことです。現時点ではよく分かりません。

コードが長いので、順番に読んでいきます。

self.server_init()

WebUI関係の関数と思われます。

self._start_target(self.targets[0])

Sessionクラスのリスト型アトリビュートtargetsを引数にとった関数です。

targetsには、Sessionのインスタンス生成時の引数targetが格納されています。 http_simple.pyの例では、Target(connection=TCPSocketConnection("127.0.0.1", 80))が格納されます。

_start_target()はTarget型の引数targetを持ち、関数内でtarget.monitorsをイテレータとしたfor文の処理のみが実装されています。

本来の_start_target()関数では監視関連の処理が実行されると思われますが、http_simple.pyの例ではインスタンスtargetの生成時に引数monitorsを指定していない（つまり、target[0].monitorsは空のリストである）ので、何も処理されないと思われます。

if self._reuse_target_connection:
    self.targets[0].open()

self.reuse_target_connectionには、Sessionインスタンス生成時のbool型引数reuse_target_connectionが格納されます。 reuse_target_connectionについては、ソースコードのコメントにて以下のように書かれています。

reuse_target_connection (bool): If True, only use one target connection instead of reconnecting each test case. Default False.

通常boofuzzがファジングを行う際はTCPコネクションの接続と切断を繰り返すので、reuse_target_connectionをTrueにした場合はファジング中のTCPコネクションを確立したままにするという事でしょうか。いずれにしろ、デフォルトではFalseなのでこのコードもスルーします。

self.num_cases_actually_fuzzed = 0
self.start_time = time.time()

num_cases_actually_fuzzedおよびstart_timeは、この行で定義されるアトリビュートです。用途はまだ分かりません。

_main_fuzz_loop()の以降のコードを理解するには_generate_mutations_indefinitely()について理解する必要がありそうなので、次回は_generate_mutations_indefinitely()から読み進めたいと思います。

satumaimoの備忘録

個人的なメモ中心

boofuzzコード読み（１）

テンプレートを見る

fuzz()のコード読み

_main_fuzz_loop()