Added pcap-comparator and fixed random seeds. the comparator still fails yet, for reasons beyond my imagination

code/Attack/BaseAttack.py

@@ -337,6 +337,7 @@ class BaseAttack(metaclass=abc.ABCMeta):
         """
         if isinstance(seed, int):
             random.seed(seed)
+            np.random.seed(seed)
 
     def set_start_time(self):
         """

code/Attack/MembersMgmtCommAttack.py

@@ -5,6 +5,7 @@ from scipy.stats import gamma
 from lea import Lea
 from datetime import datetime
 import os
+import io
 
 from Attack import BaseAttack
 from Attack.AttackParameters import Parameter as Param
@@ -252,8 +253,9 @@ class MembersMgmtCommAttack(BaseAttack.BaseAttack):
                 last_packet = packets[-1]
 
         # write the mapping to a file
-        dest_without_ext = os.path.splitext(ID2TLib.Utility.PCAP_DEST_PATH)[0]
-        msg_packet_mapping.write_to_file(dest_without_ext + "_mapping.xml")
+        buffer = io.StringIO()
+        msg_packet_mapping.write_to(buffer, close=False)
+        ID2TLib.Utility._HACKY_FILE_BUFFER["mapping.xml"] = buffer.getvalue()
 
         # Store timestamp of last packet
         self.attack_end_utime = last_packet.time

code/Core/Controller.py

@@ -25,9 +25,6 @@ class Controller:
         self.seed = None
         self.durations = []
 
-        # ugly hack, but the best we have
-        Utility.PCAP_DEST_PATH = pcap_file_path.strip()
-
         # Initialize class instances
         print("Input file: %s" % self.pcap_src_path)
         self.pcap_file = PcapFile.PcapFile(self.pcap_src_path)
@@ -110,8 +107,20 @@ class Controller:
         # write label file with attacks
         self.label_manager.write_label_file(self.pcap_dest_path)
 
+        additional_files = []
+        for suffix, content in Utility._HACKY_FILE_BUFFER.items():
+            base, _ = os.path.splitext(self.pcap_dest_path)
+            new_path = base + "_" + suffix
+
+            file = open(new_path, "w")
+            file.write(content)
+            file.close()
+            additional_files.append(new_path)
+
+        Utility._HACKY_FILE_BUFFER.clear()
+
         # print status message
-        print('\nOutput files created: \n', self.pcap_dest_path, '\n', self.label_manager.label_file_path)
+        print('\nOutput files created: \n', self.pcap_dest_path, '\n', self.label_manager.label_file_path, "\n", "\n".join(additional_files))
 
     def process_db_queries(self, query, print_results=False):
         """

code/ID2TLib/Utility.py

@@ -16,9 +16,9 @@ ROOT_DIR = CODE_DIR + "../"
 RESOURCE_DIR = ROOT_DIR + "resources/"
 TEST_DIR = RESOURCE_DIR + "test/"
 
-# this has to be initialized to None and will be set later
-# if your code needs this value and it's None something went really wrong
-PCAP_DEST_PATH = None
+# the only way for attacks to create files next to the pcap
+# todo: find a solution for this
+_HACKY_FILE_BUFFER = {}
 
 platforms = {"win7", "win10", "winxp", "win8.1", "macos", "linux", "win8", "winvista", "winnt", "win2000"}
 platform_probability = {"win7": 48.43, "win10": 27.99, "winxp": 6.07, "win8.1": 6.07, "macos": 5.94, "linux": 3.38,

code/Test/test_pcap_comparator.py

@@ -0,0 +1,154 @@
+#!/usr/bin/python3
+
+import os
+import sys
+import subprocess
+import shlex
+import time
+import unittest
+
+import scapy.all
+
+
+class PcapComparison(unittest.TestCase):
+    ID2T_DIRECTORY = ".."
+    ID2T_LOCATION = ID2T_DIRECTORY + "/" + "id2t"
+    NUM_ITERATIONS = 3
+
+    PCAP_ENVIRONMENT_VALUE = "ID2T_SRC_PCAP"
+    SEED_ENVIRONMENT_VALUE = "ID2T_SEED"
+
+    DEFAULT_PCAP = "resources/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, "-S", 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):
+            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_DIRECTORY + 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 [file.strip() for file in 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):
+        packetsA = list(scapy.all.rdpcap(self.ID2T_DIRECTORY + "/" + one))
+        packetsB = list(scapy.all.rdpcap(self.ID2T_DIRECTORY + "/" + other))
+
+        self.assertEqual(len(packetsA), len(packetsB), "Both pcap's have to have the same amount of packets")
+        for i in range(len(packetsA)):
+            p, p2 = packetsA[i], packetsB[i]
+
+            self.assertAlmostEqual(p.time, p2.time, "Packets no %i in the pcap's don't appear at the same time" % (i + 1))
+            self.compare_packets(p, p2, i + 1)
+
+    def compare_packets(self, p, p2, packet_number):
+        if p == p2:
+            return
+
+        while type(p) != scapy.packet.NoPayload or type(p2) != scapy.packet.NoPayload:
+            if type(p) != type(p2):
+                self.fail("Packets %i are of incompatible types: %s and %s" % (packet_number, type(p).__name__, type(p2).__name__))
+
+            for field in p.fields:
+                if p.fields[field] != p2.fields[field]:
+                    packet_type = type(p).__name__
+                    v, v2 = p.fields[field], p2.fields[field]
+
+                    self.fail("Packets %i differ in field %s.%s: %s != %s" %
+                                (packet_number, packet_type, field, v, v2))
+
+            p = p.payload
+            p2 = p2.payload
+
+    def print_warning(self, *text):
+        print(*text, file=sys.stderr)
+
+    def random_id2t_params(self):
+        param = lambda key, val: "-p%s=%s" % (str(key), str(val))
+
+        return [
+            []
+        ]
+
+
+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)