thrift/lib/py/util/fuzzer.py

# Copyright (c) Meta Platforms, Inc. and affiliates.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#     http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.

# pyre-unsafe

"""
Fuzz Testing for Thrift Services
"""

from __future__ import absolute_import
from __future__ import division
from __future__ import print_function
from __future__ import unicode_literals

import argparse
import collections
import inspect
import json
import logging
import os
import pprint
import sys
import time

import six
import six.moves as sm
from six.moves.urllib.parse import urlparse

try:
    # pyre-fixme[21]: Could not find module `ServiceRouter`.
    from ServiceRouter import ConnConfigs, ServiceOptions, ServiceRouter  # @manual

    SR_AVAILABLE = True
except ImportError:
    SR_AVAILABLE = False

from thrift import Thrift
from thrift.protocol import TBinaryProtocol, TCompactProtocol, THeaderProtocol
from thrift.transport import TTransport, TSocket, TSSLSocket, THttpClient
from thrift.util import randomizer

if six.PY3:
    from importlib.machinery import SourceFileLoader

    def load_source(name, pathname):
        return SourceFileLoader(name, pathname).load_module()


else:
    import imp

    def load_source(name, pathname):
        return imp.load_source(name, pathname)


def positive_int(s) -> int:
    """Typechecker for positive integers"""
    try:
        n = int(s)
        if not n > 0:
            raise argparse.ArgumentTypeError("%s is not positive." % s)
        return n
    except ValueError:
        raise argparse.ArgumentTypeError("Cannot convert %s to an integer." % s)


def prob_float(s) -> float:
    """Typechecker for probability values"""
    try:
        x = float(s)
        if not 0 <= x <= 1:
            raise argparse.ArgumentTypeError("%s is not a valid probability." % x)
        return x
    except ValueError:
        raise argparse.ArgumentTypeError("Cannot convert %s to a float." % s)


class FuzzerConfiguration(object):
    """Container for Fuzzer configuration options"""

    argspec = {
        "allow_application_exceptions": {
            "description": "Do not flag TApplicationExceptions as errors",
            "type": bool,
            "flag": "-a",
            "argparse_kwargs": {"action": "store_const", "const": True},
            "default": False,
        },
        "compact": {
            "description": "Use TCompactProtocol",
            "type": bool,
            "flag": "-c",
            "argparse_kwargs": {"action": "store_const", "const": True},
            "default": False,
        },
        "constraints": {
            "description": "JSON Constraint dictionary",
            "type": str,
            "flag": "-Con",
            "default": {},
            "is_json": True,
        },
        "framed": {
            "description": "Use framed transport.",
            "type": bool,
            "flag": "-f",
            "argparse_kwargs": {"action": "store_const", "const": True},
            "default": False,
        },
        "functions": {
            "description": "Which functions to test. If excluded, test all",
            "type": str,
            "flag": "-F",
            "argparse_kwargs": {
                "nargs": "*",
            },
            "default": None,
        },
        "host": {
            "description": "The host and port to connect to",
            "type": str,
            "flag": "-h",
            "argparse_kwargs": {"metavar": "HOST[:PORT]"},
            "default": None,
        },
        "iterations": {
            "description": "Number of calls per method.",
            "type": positive_int,
            "flag": "-n",
            "attr_name": "n_iterations",
            "default": 1000,
        },
        "logfile": {
            "description": "File to write output logs.",
            "type": str,
            "flag": "-l",
            "default": None,
        },
        "loglevel": {
            "description": "Level of verbosity to write logs.",
            "type": str,
            "flag": "-L",
            "argparse_kwargs": {
                "choices": ["DEBUG", "INFO", "WARNING", "ERROR", "CRITICAL"],
            },
            "default": "INFO",
        },
        "service": {
            "description": "Path to file of Python service module.",
            "type": str,
            "flag": "-S",
            "attr_name": "service_path",
            "default": None,
        },
        "ssl": {
            "description": "Use SSL socket.",
            "type": bool,
            "flag": "-s",
            "argparse_kwargs": {"action": "store_const", "const": True},
            "default": False,
        },
        "unframed": {
            "description": "Use unframed transport.",
            "type": bool,
            "flag": "-U",
            "argparse_kwargs": {"action": "store_const", "const": True},
            "default": False,
        },
        "url": {
            "description": "The URL to connect to for HTTP transport",
            "type": str,
            "flag": "-u",
            "default": None,
        },
    }
    if SR_AVAILABLE:
        argspec["tier"] = {
            "description": "The SMC tier to connect to",
            "type": str,
            "flag": "-t",
            "default": None,
        }
        argspec["conn_configs"] = {
            "description": "ConnConfigs to use for ServiceRouter connection",
            "type": str,
            "flag": "-Conn",
            "default": {},
            "is_json": True,
        }
        argspec["service_options"] = {
            "description": "ServiceOptions to use for ServiceRouter connection",
            "type": str,
            "flag": "-SO",
            "default": {},
            "is_json": True,
        }

    def __init__(self, service=None):
        cls = self.__class__

        if service is not None:
            self.service = service

        parser = argparse.ArgumentParser(
            description="Fuzzer Configuration", add_help=False
        )
        parser.add_argument(
            "-C",
            "--config",
            dest="config_filename",
            help="JSON Configuration file. "
            "All settings can be specified as commandline "
            "args and config file settings. Commandline args "
            "override config file settings.",
        )

        parser.add_argument(
            "-?", "--help", action="help", help="Show this help message and exit."
        )

        for name, arg in six.iteritems(cls.argspec):
            kwargs = arg.get("argparse_kwargs", {})

            if kwargs.get("action", None) != "store_const":
                # Pass type to argparse. With store_const, type can be inferred
                kwargs["type"] = arg["type"]

            # If an argument is not passed, don't put a value in the namespace
            kwargs["default"] = argparse.SUPPRESS

            # Use the argument's description and default as a help message
            kwargs["help"] = "%s Default: %s" % (
                arg.get("description", ""),
                arg["default"],
            )

            kwargs["dest"] = arg.get("attr_name", name)

            if hasattr(self, kwargs["dest"]):
                # Attribute already assigned (e.g., service passed to __init__)
                continue

            parser.add_argument(arg["flag"], "--%s" % name, **kwargs)

            # Assign the default value to config namespace
            setattr(self, kwargs["dest"], arg["default"])

        args = parser.parse_args()

        # Read settings in config file
        self.__dict__.update(cls._config_file_settings(args))

        # Read settings in args
        self.__dict__.update(cls._args_settings(args))

        valid, message = self._validate_config()
        if not valid:
            print(message, file=sys.stderr)
            sys.exit(os.EX_USAGE)

    @classmethod
    def _try_parse_type(cls, name, type_, val):
        try:
            val = type_(val)
        except ValueError:
            raise TypeError(
                ("Expected type %s for setting %s, " "but got type %s (%s)")
                % (type_, name, type(val), val)
            )
        return val

    @classmethod
    def _try_parse(cls, name, arg, val):
        if arg.get("is_json", False):
            return val

        type_ = arg["type"]

        nargs = arg.get("argparse_kwargs", {}).get("nargs", None)

        if nargs is None:
            return cls._try_parse_type(name, type_, val)
        else:
            if not isinstance(val, list):
                raise TypeError(
                    (
                        "Expected list of length %s "
                        "for setting %s, but got type %s (%s)"
                    )
                    % (nargs, name, type(val), val)
                )
            ret = []
            for elem in val:
                ret.append(cls._try_parse_type(name, type_, elem))
            return ret

    @classmethod
    def _config_file_settings(cls, args):
        """Read settings from a configuration file"""
        if args.config_filename is None:
            return {}  # No config file
        if not os.path.exists(args.config_filename):
            raise OSError(
                os.EX_NOINPUT, "Config file does not exist: %s" % args.config_filename
            )
        with open(args.config_filename, "r") as fd:
            try:
                settings = json.load(fd)
            except ValueError as e:
                raise ValueError("Error parsing config file: %s" % e)

        # Make sure settings are well-formatted
        renamed_settings = {}
        if not isinstance(settings, dict):
            raise TypeError("Invalid config file. Top-level must be Object.")
        for name, val in six.iteritems(settings):
            if name not in cls.argspec:
                raise ValueError(("Unrecognized configuration " "option: %s") % name)
            arg = cls.argspec[name]
            val = cls._try_parse(name, arg, val)
            attr_name = arg.get("attr_name", name)
            renamed_settings[attr_name] = val
        return renamed_settings

    @classmethod
    def _args_settings(cls, args):
        """Read settings from the args namespace returned by argparse"""
        settings = {}
        for name, arg in six.iteritems(cls.argspec):
            attr_name = arg.get("attr_name", name)
            if not hasattr(args, attr_name):
                continue
            value = getattr(args, attr_name)
            if arg.get("is_json", False):
                settings[attr_name] = json.loads(value)
            else:
                settings[attr_name] = value
        return settings

    def __str__(self):
        return "Configuration(\n%s\n)" % pprint.pformat(self.__dict__)

    def load_service(self):
        if self.service is not None:
            if self.service_path is not None:
                raise ValueError(
                    "Cannot specify a service path when the "
                    "service is input programmatically"
                )
            # Service already loaded programmatically. Just load methods.
            self.service.load_methods()
            return self.service

        if self.service_path is None:
            raise ValueError("Error: No service specified")

        service_path = self.service_path

        if not os.path.exists(service_path):
            raise OSError("Service module does not exist: %s" % service_path)

        if not service_path.endswith(".py"):
            raise OSError("Service module is not a Python module: %s" % service_path)

        parent_path, service_filename = os.path.split(service_path)
        service_name = service_filename[:-3]  # Truncate extension

        logging.info("Service name: %s" % (service_name))

        parent_path = os.path.dirname(service_path)
        ttypes_path = os.path.join(parent_path, "ttypes.py")
        constants_path = os.path.join(parent_path, "constants.py")

        load_source("module", parent_path)
        ttypes_module = load_source("module.ttypes", ttypes_path)
        constants_module = load_source("module.constants", constants_path)
        service_module = load_source("module.%s" % (service_name), service_path)

        service = Service(ttypes_module, constants_module, service_module)
        service.load_methods()
        return service

    def _validate_config(self):
        # Verify there is one valid connection flag
        specified_flags = []
        connection_flags = FuzzerClient.connection_flags
        for flag in connection_flags:
            if hasattr(self, flag) and getattr(self, flag) is not None:
                specified_flags.append(flag)

        if not len(specified_flags) == 1:
            message = "Exactly one of [%s] must be specified. Got [%s]." % (
                (", ".join("--%s" % flag for flag in connection_flags)),
                (", ".join("--%s" % flag for flag in specified_flags)),
            )
            return False, message

        connection_method = specified_flags[0]
        self.connection_method = connection_method

        if connection_method == "url":
            if not (self.compact or self.framed or self.unframed):
                message = (
                    "A protocol (compact, framed, or unframed) "
                    "must be specified for HTTP Transport."
                )
                return False, message

        if connection_method in {"url", "host"}:
            if connection_method == "url":
                try:
                    url = urlparse(self.url)
                except Exception:
                    return False, "Unable to parse url %s" % self.url
                else:
                    connection_str = url[1]
            elif connection_method == "host":
                connection_str = self.host
            if ":" in connection_str:
                # Get the string after the colon
                port = connection_str[connection_str.index(":") + 1 :]
                try:
                    int(port)
                except ValueError:
                    message = "Port is not an integer: %s" % port
                    return False, message

        return True, None


class Service(object):
    """Wrapper for a thrift service"""

    def __init__(self, ttypes_module, constants_module, service_module):
        self.ttypes = ttypes_module
        self.constants = constants_module
        self.service = service_module
        self.methods = None

    def __str__(self):
        return "Service(%s)" % self.service.__name__

    def load_methods(self, exclude_ifaces=None):
        """Load a service's methods.

        If exclude_ifaces is not None, it should be a collection and only
        methods from thrift interfaces not included in that collection will
        be considered."""

        exclude_ifaces = exclude_ifaces or []

        pred = inspect.isfunction if six.PY3 else inspect.ismethod

        methods = {}
        exclude_methods = []

        for klass in exclude_ifaces:
            exclude_methods.extend(inspect.getmembers(klass, predicate=pred))

        klass_methods = inspect.getmembers(self.service.Iface, predicate=pred)

        for method_name, method in klass_methods:
            if (method_name, method) in exclude_methods:
                continue

            module = inspect.getmodule(method)

            args = getattr(module, method_name + "_args", None)
            if args is None:
                continue
            result = getattr(module, method_name + "_result", None)

            thrift_exceptions = []
            if result is not None:
                for res_spec in result.thrift_spec:
                    if res_spec is None:
                        continue
                    if res_spec[2] != "success":
                        # This is an exception return type
                        spec_args = res_spec[3]
                        exception_type = spec_args[0]
                        thrift_exceptions.append(exception_type)

            methods[method_name] = {
                "args_class": args,
                "result_spec": result,
                "thrift_exceptions": tuple(thrift_exceptions),
            }

        self.methods = methods

    @property
    def client_class(self):
        return self.service.Client

    def get_methods(self, include=None):
        """Get a dictionary of methods provided by the service.

        If include is not None, it should be a collection and only
        the method names in that collection will be included."""

        if self.methods is None:
            raise ValueError(
                "Service.load_methods must be " "called before Service.get_methods"
            )

        if include is None:
            return self.methods

        included_methods = {}
        for method_name in include:
            if method_name not in self.methods:
                raise NameError("Function does not exist: %s" % method_name)
            included_methods[method_name] = self.methods[method_name]

        return included_methods


class FuzzerClient(object):
    """Client wrapper used to make calls based on configuration settings"""

    connection_flags = ["host", "url", "tier"]
    default_port = 9090

    def __init__(self, config, client_class):
        self.config = config
        self.client_class = client_class

    def _get_client_by_transport(self, config, transport, socket=None):
        # Create the protocol and client
        if config.compact:
            protocol = TCompactProtocol.TCompactProtocol(transport)
        # No explicit option about protocol is specified. Try to infer.
        elif config.framed or config.unframed:
            protocol = TBinaryProtocol.TBinaryProtocolAccelerated(transport)
        elif socket is not None:
            protocol = THeaderProtocol.THeaderProtocol(socket)
            transport = protocol.trans
        else:
            raise ValueError("No protocol specified for HTTP Transport")
        transport.open()
        self._transport = transport

        client = self.client_class(protocol)
        return client

    def _parse_host_port(self, value, default_port):
        parts = value.rsplit(":", 1)
        if len(parts) == 1:
            return (parts[0], default_port)
        else:
            # FuzzerConfiguration ensures parts[1] is an int
            return (parts[0], int(parts[1]))

    def _get_client_by_host(self):
        config = self.config
        host, port = self._parse_host_port(config.host, self.default_port)
        socket = (
            TSSLSocket.TSSLSocket(host, port)
            if config.ssl
            else TSocket.TSocket(host, port)
        )
        if config.framed:
            transport = TTransport.TFramedTransport(socket)
        else:
            transport = TTransport.TBufferedTransport(socket)
        return self._get_client_by_transport(config, transport, socket=socket)

    def _get_client_by_url(self):
        config = self.config
        url = urlparse(config.url)
        host, port = self._parse_host_port(url[1], 80)
        transport = THttpClient.THttpClient(config.url)
        return self._get_client_by_transport(config, transport)

    def _get_client_by_tier(self):
        """Get a client that uses ServiceRouter"""
        config = self.config
        serviceRouter = ServiceRouter()

        overrides = ConnConfigs()
        for key, val in six.iteritems(config.conn_configs):
            key = six.binary_type(key)
            val = six.binary_type(val)
            overrides[key] = val

        sr_options = ServiceOptions()
        for key, val in six.iteritems(config.service_options):
            key = six.binary_type(key)
            if not isinstance(val, list):
                raise TypeError(
                    "Service option %s expected list; got %s (%s)"
                    % (key, val, type(val))
                )
            val = [six.binary_type(elem) for elem in val]
            sr_options[key] = val

        service_name = config.tier

        # Obtain a normal client connection using SR2
        client = serviceRouter.getClient2(
            self.client_class, service_name, sr_options, overrides
        )

        if client is None:
            raise NameError("Failed to lookup host for tier %s" % service_name)

        return client

    def _get_client(self):
        if self.config.connection_method == "host":
            client = self._get_client_by_host()
        elif self.config.connection_method == "url":
            client = self._get_client_by_url()
        elif self.config.connection_method == "tier":
            client = self._get_client_by_tier()
        else:
            raise NameError(
                "Unknown connection type: %s" % self.config.connection_method
            )
        return client

    def _close_client(self):
        if self.config.connection_method in {"host", "url"}:
            self._transport.close()

    def __enter__(self):
        self.client = self._get_client()
        return self

    def __exit__(self, exc_type, exc_value, traceback):
        self._close_client()
        self.client = None

    def reset(self):
        self._close_client()
        try:
            self.client = self._get_client()
            return True
        except TTransport.TTransportException as e:
            logging.error("Unable to reset connection: %r" % e)
            return False

    def make_call(self, method_name, kwargs, is_oneway=False):
        method = getattr(self.client, method_name)
        ret = method(**kwargs)

        if is_oneway:
            self.reset()

        return ret


class Timer(object):
    def __init__(self, aggregator, category, action):
        self.aggregator = aggregator
        self.category = category
        self.action = action

    def __enter__(self):
        self.start_time = time.time()

    def __exit__(self, exc_type, exc_value, traceback):
        end_time = time.time()
        time_elapsed = end_time - self.start_time
        self.aggregator.add(self.category, self.action, time_elapsed)


class TimeAggregator(object):
    def __init__(self):
        self.total_time = collections.defaultdict(
            lambda: collections.defaultdict(float)
        )

    def time(self, category, action):
        return Timer(self, category, action)

    def add(self, category, action, time_elapsed):
        self.total_time[category][action] += time_elapsed

    def summarize(self):
        max_category_name_length = max(len(name) for name in self.total_time)
        max_action_name_length = max(
            max(len(action_name) for action_name in self.total_time[name])
            for name in self.total_time
        )
        category_format = "%%%ds: %%s" % max_category_name_length
        action_format = "%%%ds: %%4.3fs" % max_action_name_length

        category_summaries = []
        for category_name, category_actions in sorted(self.total_time.items()):
            timing_items = []
            for action_name, action_time in sorted(category_actions.items()):
                timing_items.append(action_format % (action_name, action_time))
            all_actions = " | ".join(timing_items)
            category_summaries.append(category_format % (category_name, all_actions))
        summaries = "\n".join(category_summaries)
        logging.info("Timing Summary:\n%s" % summaries)


class FuzzTester(object):
    summary_interval = 1  # Seconds between summary logs

    class Result:
        Success = 0
        TransportException = 1
        ApplicationException = 2
        UserDefinedException = 3
        OtherException = 4
        Crash = 5

    def __init__(self, config):
        self.config = config
        self.service = None
        self.randomizer = None
        self.client = None

    def start_logging(self):
        logfile = self.config.logfile
        if self.config.logfile is None:
            logfile = "/dev/null"
        log_level = getattr(logging, self.config.loglevel)

        datefmt = "%Y-%m-%d %H:%M:%S"
        fmt = "[%(asctime)s] [%(levelname)s] %(message)s"

        if logfile == "stdout":
            logging.basicConfig(stream=sys.stdout, level=log_level)
        else:
            logging.basicConfig(filename=self.config.logfile, level=log_level)

        log_handler = logging.getLogger().handlers[0]
        log_handler.setFormatter(logging.Formatter(fmt, datefmt=datefmt))

    def start_timing(self):
        self.timer = TimeAggregator()
        self.next_summary_time = time.time() + self.__class__.summary_interval

    def _call_string(self, method_name, kwargs):
        kwarg_str = ", ".join("%s=%s" % (k, v) for k, v in six.iteritems(kwargs))
        return "%s(%s)" % (method_name, kwarg_str)

    def run_test(
        self, method_name, kwargs, expected_output, is_oneway, thrift_exceptions
    ):
        """
        Make an RPC with given arguments and check for exceptions.
        """
        try:
            with self.timer.time(method_name, "Thrift"):
                self.client.make_call(method_name, kwargs, is_oneway)
        except thrift_exceptions as e:
            self.record_result(method_name, FuzzTester.Result.UserDefinedException)
            if self.config.loglevel == "DEBUG":
                with self.timer.time(method_name, "Logging"):
                    logging.debug("Got thrift exception: %r" % e)
                    logging.debug(
                        "Exception thrown by call: %s"
                        % (self._call_string(method_name, kwargs))
                    )

        except Thrift.TApplicationException as e:
            self.record_result(method_name, FuzzTester.Result.ApplicationException)
            if self.config.allow_application_exceptions:
                if self.config.loglevel == "DEBUG":
                    with self.timer.time(method_name, "Logging"):
                        logging.debug("Got TApplication exception %s" % e)
                        logging.debug(
                            "Exception thrown by call: %s"
                            % (self._call_string(method_name, kwargs))
                        )
            else:
                with self.timer.time(method_name, "Logging"):
                    self.n_exceptions += 1
                    logging.error("Got application exception: %s" % e)
                    logging.error(
                        "Offending call: %s" % (self._call_string(method_name, kwargs))
                    )

        except TTransport.TTransportException as e:
            self.n_exceptions += 1

            with self.timer.time(method_name, "Logging"):
                logging.error("Got TTransportException: (%s, %r)" % (e, e))
                logging.error(
                    "Offending call: %s" % (self._call_string(method_name, kwargs))
                )

            if "errno = 111: Connection refused" in e.args[0]:
                # Unable to connect to server - server may be down
                self.record_result(method_name, FuzzTester.Result.Crash)
                return False

            if not self.client.reset():
                logging.error("Inferring server crash.")
                self.record_result(method_name, FuzzTester.Result.Crash)
                return False

            self.record_result(method_name, FuzzTester.Result.TransportException)

        except Exception as e:
            self.record_result(method_name, FuzzTester.Result.OtherException)
            with self.timer.time(method_name, "Logging"):
                self.n_exceptions += 1
                logging.error("Got exception %s (%r)" % (e, e))
                logging.error(
                    "Offending call: %s" % (self._call_string(method_name, kwargs))
                )
                if hasattr(self, "previous_kwargs"):
                    logging.error(
                        "Previous call: %s"
                        % (self._call_string(method_name, self.previous_kwargs))
                    )

        else:
            self.record_result(method_name, FuzzTester.Result.Success)
            if self.config.loglevel == "DEBUG":
                with self.timer.time(method_name, "Logging"):
                    logging.debug(
                        "Successful call: %s" % (self._call_string(method_name, kwargs))
                    )
        finally:
            self.n_tests += 1

        return True

    def fuzz_kwargs(self, method_name, n_iterations):
        # For now, just yield n random sets of args
        # In future versions, fuzz fields more methodically based
        # on feedback and seeds
        for _ in sm.xrange(n_iterations):
            with self.timer.time(method_name, "Randomizing"):
                method_randomizer = self.method_randomizers[method_name]
                args_struct = method_randomizer.generate()
            if args_struct is None:
                logging.error("Unable to produce valid arguments for %s" % method_name)
            else:
                kwargs = args_struct.__dict__  # Get members of args struct
                yield kwargs

    def get_method_randomizers(self, methods, constraints):
        """Create a StructRandomizer for each method"""
        state = randomizer.RandomizerState()
        method_randomizers = {}

        state.push_type_constraints(constraints)

        for method_name in methods:
            method_constraints = constraints.get(method_name, {})
            args_class = methods[method_name]["args_class"]

            # Create a spec_args tuple for the method args struct type
            randomizer_spec_args = (
                args_class,
                args_class.thrift_spec,
                False,  # isUnion
            )

            method_randomizer = state.get_randomizer(
                Thrift.TType.STRUCT, randomizer_spec_args, method_constraints
            )
            method_randomizers[method_name] = method_randomizer

        return method_randomizers

    def _split_key(self, key):
        """Split a constraint rule key such as a.b|c into ['a', 'b', '|c']
        Dots separate hierarchical field names and property names

        Pipes indicate a type name and hashes indicate a field name,
        though these rules are not yet supported.
        """
        components = []
        start_idx = 0
        cur_idx = 0
        while cur_idx < len(key):
            if cur_idx != start_idx and key[cur_idx] in {".", "|", "#"}:
                components.append(key[start_idx:cur_idx])
                start_idx = cur_idx
                if key[cur_idx] == ".":
                    start_idx += 1
                cur_idx = start_idx
            else:
                cur_idx += 1
        components.append(key[start_idx:])
        return components

    def preprocess_constraints(self, source_constraints):
        """
        The constraints dictionary can have any key
        that follows the following format:

        method_name[.arg_name][.field_name ...].property_name

        The values in the dictionary can be nested such that inner field
        names are subfields of the outer scope, and inner type rules are
        applied only to subvalues of the out scope.

        After preprocessing, each dictionary level should have exactly one
        method name, field name, or property name as its key.

        Any strings of identifiers are converted into the nested dictionary
        structure. For example, the constraint set:

        {'my_method.my_field.distribution': 'uniform(0,100)'}

        Will be preprocessed to:

        {'my_method':
          {'my_field':
             {'distribution': 'uniform(0, 100)'}
          }
        }
        """
        constraints = {}
        scope_path = []

        def add_constraint(rule):
            walk_scope = constraints
            for key in scope_path[:-1]:
                if key not in walk_scope:
                    walk_scope[key] = {}
                walk_scope = walk_scope[key]
            walk_scope[scope_path[-1]] = rule

        def add_constraints_from_dict(d):
            for key, rule in six.iteritems(d):
                key_components = self._split_key(key)
                scope_path.extend(key_components)
                if isinstance(rule, dict):
                    add_constraints_from_dict(rule)
                else:
                    add_constraint(rule)
                scope_path[-len(key_components) :] = []

        add_constraints_from_dict(source_constraints)
        return constraints

    def start_result_counters(self):
        """Create result counters. The counters object is a dict that maps
        a method name to a counter of FuzzTest.Results
        """
        self.result_counters = collections.defaultdict(collections.Counter)

    def record_result(self, method_name, result):
        self.result_counters[method_name][result] += 1

    def log_result_summary(self, method_name):
        if time.time() >= self.next_summary_time:
            results = []
            for name, val in six.iteritems(vars(FuzzTester.Result)):
                if name.startswith("_"):
                    continue
                count = self.result_counters[method_name][val]
                if count > 0:
                    results.append((name, count))
            results.sort()
            logging.info(
                "%s count: {%s}"
                % (method_name, ", ".join("%s: %d" % r for r in results))
            )

            interval = self.__class__.summary_interval
            # Determine how many full intervals have passed between
            # self.next_summary_time (the scheduled time for this summary) and
            # the time the summary is actually completed.
            intervals_passed = int((time.time() - self.next_summary_time) / interval)
            # Schedule the next summary for the first interval that has not yet
            # fully passed
            self.next_summary_time += interval * (intervals_passed + 1)

    def run(self):
        self.start_logging()
        self.start_timing()
        self.start_result_counters()

        logging.info("Starting Fuzz Tester")
        logging.info(str(self.config))

        self.service = self.config.load_service()

        client_class = self.service.client_class

        methods = self.service.get_methods(self.config.functions)
        constraints = self.preprocess_constraints(self.config.constraints)
        self.method_randomizers = self.get_method_randomizers(methods, constraints)

        logging.info("Fuzzing methods: %s" % methods.keys())

        with FuzzerClient(self.config, client_class) as self.client:
            for method_name, spec in six.iteritems(methods):
                result_spec = spec.get("result_spec", None)
                thrift_exceptions = spec["thrift_exceptions"]
                is_oneway = result_spec is None
                logging.info("Fuzz testing method %s" % (method_name))
                self.n_tests = 0
                self.n_exceptions = 0
                did_crash = False
                for kwargs in self.fuzz_kwargs(method_name, self.config.n_iterations):
                    if not self.run_test(
                        method_name, kwargs, None, is_oneway, thrift_exceptions
                    ):
                        did_crash = True
                        break
                    self.log_result_summary(method_name)
                    self.previous_kwargs = kwargs

                if did_crash:
                    logging.error(
                        ("Method %s caused the " "server to crash.") % (method_name)
                    )
                    break
                else:
                    logging.info(
                        ("Method %s raised unexpected " "exceptions in %d/%d tests.")
                        % (method_name, self.n_exceptions, self.n_tests)
                    )

        self.timer.summarize()


def run_fuzzer(config) -> None:
    fuzzer = FuzzTester(config)
    fuzzer.run()


def fuzz_service(service: Service, ttypes, constants) -> None:
    """Run the tester with required modules input programmatically"""
    service = Service(ttypes, constants, service)
    config = FuzzerConfiguration(service)
    run_fuzzer(config)


if __name__ == "__main__":
    config = FuzzerConfiguration()
    run_fuzzer(config)