ID2T-toolkit/test/test_pcap_comparator.py

#!/usr/bin/python3

import sys, os
import subprocess, shlex
import time
import unittest
import random

import scapy.all
from TestUtil import PcapComparator

# this dictionary holds the generators (functions) for the parameters
# that will be passed to the MembershipMgmtCommAttack
# items need the parameter-name as key and a function that will be called
# without parameters and returns a valid value for that parameter as value
# WARNING: parameters will be passed via command line, make sure your values
# get converted to string correctly
_random_bool = lambda: random.random() < 0.5
ID2T_PARAMETER_GENERATORS = {
    "bots.count": lambda: random.randint(3, 6),
#    "file.csv":,
#    "file.xml":,
    "hidden_mark": _random_bool,
#    "interval.selection.end":,
#    "interval.selection.start":,
#    "interval.selection.strategy":,
#    "ip.reuse.external":,
#    "ip.reuse.local":,
#    "ip.reuse.total":,
    "multiport": _random_bool,
    "nat.present": _random_bool,
    "packet.padding": lambda: random.randint(0, 100),
    "packets.limit": lambda: random.randint(50, 150),
    "packets.per-second": lambda: random.randint(1000, 2000) / 100,
    "ttl.from.caida": _random_bool,
}

class PcapComparison(unittest.TestCase):
    ID2T_PATH = ".."
    ID2T_LOCATION = ID2T_PATH + "/" + "id2t"
    NUM_ITERATIONS_PER_PARAMS = 3
    NUM_ITERATIONS = 5

    PCAP_ENVIRONMENT_VALUE = "ID2T_SRC_PCAP"
    SEED_ENVIRONMENT_VALUE = "ID2T_SEED"

    DEFAULT_PCAP = "resources/test/telnet-raw.pcap"
    DEFAULT_SEED = "42"

    OUTPUT_FILES_PREFIX_LINE = "Output files created:"

    def __init__(self, *args, **kwargs):
        unittest.TestCase.__init__(self, *args, **kwargs)

        # params to call id2t with, as a list[list[str]]
        # do a round of testing for each list[str] we get
        # if none generate some params itself
        self.id2t_params = None

    def set_id2t_params(self, params: "list[list[str]]"):
        self.id2t_params = params

    def setUp(self):
        self.generated_files = []
        self.keep_files = []

    def test_determinism(self):
        input_pcap = os.environ.get(self.PCAP_ENVIRONMENT_VALUE, self.DEFAULT_PCAP)
        seed = os.environ.get(self.SEED_ENVIRONMENT_VALUE, self.DEFAULT_SEED)

        if self.id2t_params is None:
            self.id2t_params = self.random_id2t_params()

        for params in self.id2t_params:
            self.do_test_round(input_pcap, seed, params)

    def do_test_round(self, input_pcap, seed, additional_params):
        command_args = [self.ID2T_LOCATION, "-i", input_pcap, "--seed", seed, "-a", "MembersMgmtCommAttack"] + additional_params
        command = " ".join(map(shlex.quote, command_args))
        self.print_warning("The command that gets executed is:", command)

        generated_pcap = None
        for i in range(self.NUM_ITERATIONS_PER_PARAMS):
            retcode, output = subprocess.getstatusoutput(command)

            self.print_warning(output)
            self.assertEqual(retcode, 0, "For some reason id2t completed with an error")

            files = self.parse_files(output)
            self.generated_files.extend(files)

            pcap = self.find_pcap(files)
            if generated_pcap is not None:
                try:
                    self.compare_pcaps(generated_pcap, pcap)
                except AssertionError as e:
                    self.keep_files = [generated_pcap, pcap]
                    raise e
            else:
                generated_pcap = pcap

            self.print_warning()
            time.sleep(1)  # let some time pass between calls because files are based on the time

    def tearDown(self):
        self.print_warning("Cleaning up files generated by the test-calls...")
        for file in self.generated_files:
            if file in self.keep_files: continue

            self.print_warning(file)
            os.remove(self.ID2T_PATH + os.path.sep + file)
        self.print_warning("Done")
        self.print_warning("The following files have been kept: " + ", ".join(self.keep_files))

    def parse_files(self, program_output: str) -> "list[str]":
        lines = program_output.split(os.linesep)

        self.assertIn(self.OUTPUT_FILES_PREFIX_LINE, lines,
                "The magic string is not in the program output anymore, has the program output structure changed?")
        index = lines.index(self.OUTPUT_FILES_PREFIX_LINE)

        return lines[index + 1:]

    def find_pcap(self, files: "list[str]") -> str:
        return next(file for file in files if file.endswith(".pcap"))

    def compare_pcaps(self, one: str, other: str):
        PcapComparator().compare_files(self.ID2T_PATH + "/" + one, self.ID2T_PATH + "/" + other)

    def print_warning(self, *text):
        print(*text, file=sys.stderr)

    def random_id2t_params(self):
        """
        :return: A list of parameter-lists for id2t, useful if you want several
        iterations
        """
        param_list = []
        for i in range(self.NUM_ITERATIONS):
            param_list.append(self.random_id2t_param_set())
        return param_list

    def random_id2t_param_set(self):
        """
        Create a list of parameters to call the membersmgmtcommattack with
        :return: a list of command-line parameters
        """
        param = lambda key, val: "%s=%s" % (str(key), str(val))

        number_of_keys = min(random.randint(2, 5), len(ID2T_PARAMETER_GENERATORS))
        keys = random.sample(list(ID2T_PARAMETER_GENERATORS), number_of_keys)

        params = []
        for key in keys:
            generator = ID2T_PARAMETER_GENERATORS[key]
            params.append(param(key, generator()))

        return params

if __name__ == "__main__":
    import sys

    # parameters for this program are interpreted as id2t-parameters
    id2t_args = sys.argv[1:]
    comparison = PcapComparison("test_determinism")
    if id2t_args: comparison.set_id2t_params([id2t_args])

    suite = unittest.TestSuite()
    suite.addTest(comparison)

    unittest.TextTestRunner().run(suite)