ClickHouse/tests/fuzz/runner.py

#!/usr/bin/env python3

import configparser
import logging
import os
import re
import signal
import subprocess
from pathlib import Path
from time import sleep
from typing import List

from botocore.exceptions import ClientError

DEBUGGER = os.getenv("DEBUGGER", "")
FUZZER_ARGS = os.getenv("FUZZER_ARGS", "")


def report(source: str, reason: str, call_stack: list, test_unit: str):
    logging.info("########### REPORT: %s %s %s", source, reason, test_unit)
    logging.info("".join(call_stack))
    logging.info("########### END OF REPORT ###########")


# pylint: disable=unused-argument
def process_fuzzer_output(output: str):
    pass


def process_error(error: str) -> list:
    ERROR = r"^==\d+==\s?ERROR: (\S+): (.*)"
    error_source = ""
    error_reason = ""
    test_unit = ""
    TEST_UNIT_LINE = r"artifact_prefix='.*\/'; Test unit written to (.*)"
    error_info = []
    is_error = False

    # pylint: disable=unused-variable
    for line_num, line in enumerate(error.splitlines(), 1):
        if is_error:
            error_info.append(line)
            match = re.search(TEST_UNIT_LINE, line)
            if match:
                test_unit = match.group(1)
            continue

        match = re.search(ERROR, line)
        if match:
            error_info.append(line)
            error_source = match.group(1)
            error_reason = match.group(2)
            is_error = True

    report(error_source, error_reason, error_info, test_unit)
    return error_info


def kill_fuzzer(fuzzer: str):
    with subprocess.Popen(["ps", "-A", "u"], stdout=subprocess.PIPE) as p:
        out, _ = p.communicate()
        for line in out.splitlines():
            if fuzzer.encode("utf-8") in line:
                pid = int(line.split(None, 2)[1])
                logging.info("Killing fuzzer %s, pid %d", fuzzer, pid)
                os.kill(pid, signal.SIGKILL)


def run_fuzzer(fuzzer: str, timeout: int) -> TestResult:
    s3 = S3Helper()

    logging.info("Running fuzzer %s...", fuzzer)

    seed_corpus_dir = f"{fuzzer}.in"
    with Path(seed_corpus_dir) as path:
        if not path.exists() or not path.is_dir():
            seed_corpus_dir = ""

    active_corpus_dir = f"{fuzzer}.corpus"
    try:
        s3.download_files(
            bucket=S3_BUILDS_BUCKET,
            s3_path=f"fuzzer/corpus/{fuzzer}/",
            file_suffix="",
            local_directory=active_corpus_dir,
        )
    except ClientError as e:
        if e.response["Error"]["Code"] == "NoSuchKey":
            logging.debug("No active corpus exists for %s", fuzzer)
        else:
            raise

    new_corpus_dir = f"{fuzzer}.corpus_new"
    if not os.path.exists(new_corpus_dir):
        os.makedirs(new_corpus_dir)

    options_file = f"{fuzzer}.options"
    custom_libfuzzer_options = ""
    fuzzer_arguments = ""

    with Path(options_file) as path:
        if path.exists() and path.is_file():
            parser = configparser.ConfigParser()
            parser.read(path)

            if parser.has_section("asan"):
                os.environ["ASAN_OPTIONS"] = (
                    f"{os.environ['ASAN_OPTIONS']}:{':'.join(f'{key}={value}' for key, value in parser['asan'].items())}"
                )

            if parser.has_section("msan"):
                os.environ["MSAN_OPTIONS"] = (
                    f"{os.environ['MSAN_OPTIONS']}:{':'.join(f'{key}={value}' for key, value in parser['msan'].items())}"
                )

            if parser.has_section("ubsan"):
                os.environ["UBSAN_OPTIONS"] = (
                    f"{os.environ['UBSAN_OPTIONS']}:{':'.join(f'{key}={value}' for key, value in parser['ubsan'].items())}"
                )

            if parser.has_section("libfuzzer"):
                custom_libfuzzer_options = " ".join(
                    f"-{key}={value}"
                    for key, value in parser["libfuzzer"].items()
                    if key != "jobs"
                )

            if parser.has_section("fuzzer_arguments"):
                fuzzer_arguments = " ".join(
                    (f"{key}") if value == "" else (f"{key}={value}")
                    for key, value in parser["fuzzer_arguments"].items()
                )

    cmd_line = f"{DEBUGGER} ./{fuzzer} {FUZZER_ARGS} {new_corpus_dir} {active_corpus_dir} {seed_corpus_dir}"

    if custom_libfuzzer_options:
        cmd_line += f" {custom_libfuzzer_options}"
    if fuzzer_arguments:
        cmd_line += f" {fuzzer_arguments}"

    if not "-dict=" in cmd_line and Path(f"{fuzzer}.dict").exists():
        cmd_line += f" -dict={fuzzer}.dict"

    cmd_line += " < /dev/null"

    logging.info("...will execute: %s", cmd_line)

    test_result = TestResult(fuzzer, "OK")
    stopwatch = Stopwatch()
    try:
        result = subprocess.run(
            cmd_line,
            stderr=subprocess.PIPE,
            stdout=subprocess.DEVNULL,
            text=True,
            check=True,
            shell=True,
            errors="replace",
            timeout=timeout,
        )
    except subprocess.CalledProcessError as e:
        # print("Command failed with error:", e)
        logging.info("Stderr output: %s", e.stderr)
        test_result = TestResult(
            fuzzer,
            "FAIL",
            stopwatch.duration_seconds,
            "",
            "\n".join(process_error(e.stderr)),
        )
    except subprocess.TimeoutExpired as e:
        logging.info("Timeout for %s", cmd_line)
        kill_fuzzer(fuzzer)
        sleep(10)
        process_fuzzer_output(e.stderr)
        test_result = TestResult(
            fuzzer,
            "Timeout",
            stopwatch.duration_seconds,
            "",
            "",
        )
    else:
        process_fuzzer_output(result.stderr)
        test_result.time = stopwatch.duration_seconds

    s3.upload_build_directory_to_s3(
        Path(new_corpus_dir), f"fuzzer/corpus/{fuzzer}", False
    )

    logging.info("test_result: %s", test_result)
    return test_result


def main():
    logging.basicConfig(level=logging.INFO)

    subprocess.check_call("ls -al", shell=True)

    timeout = 30

    match = re.search(r"(^|\s+)-max_total_time=(\d+)($|\s)", FUZZER_ARGS)
    if match:
        timeout += int(match.group(2))

    test_results = []
    stopwatch = Stopwatch()
    with Path() as current:
        for fuzzer in current.iterdir():
            if (current / fuzzer).is_file() and os.access(current / fuzzer, os.X_OK):
                test_results.append(run_fuzzer(fuzzer.name, timeout))

    prepared_results = prepare_tests_results_for_clickhouse(PRInfo(), test_results, "failure", stopwatch.duration_seconds, stopwatch.start_time_str, "", "libFuzzer")
    # ch_helper = ClickHouseHelper()
    # ch_helper.insert_events_into(db="default", table="checks", events=prepared_results)
    logging.info("prepared_results: %s", prepared_results)


if __name__ == "__main__":
    from os import path, sys

    ACTIVE_DIR = path.dirname(path.abspath(__file__))
    sys.path.append((Path(path.dirname(ACTIVE_DIR)) / "ci").as_posix())
    from env_helper import (  # pylint: disable=import-error,no-name-in-module
        S3_BUILDS_BUCKET,
    )
    from s3_helper import S3Helper  # pylint: disable=import-error,no-name-in-module
    from clickhouse_helper import (  # pylint: disable=import-error,no-name-in-module
        ClickHouseHelper,
        prepare_tests_results_for_clickhouse,
    )
    from pr_info import PRInfo  # pylint: disable=import-error,no-name-in-module
    from stopwatch import Stopwatch  # pylint: disable=import-error,no-name-in-module
    from report import TestResult  # pylint: disable=import-error,no-name-in-module

    main()