Merge branch 'unittest_squashing' of stefan.schmidt/ID2T-toolkit into master

.gitignore

@@ -20,6 +20,8 @@ tures/*
 
 code/*.pcap
 code/*.xml
+.coverage
+htmlcov/
 dbs/
 *.csv
 *.stat
@@ -27,3 +29,6 @@ dbs/
 code_boost/src/cxx/CMakeLists.txt
 code_boost/src/cxx/cmake-build-debug/
 code_boost/src/cmake-build-debug/
+resources/test/ID2T_results
+
+run_tests

build.sh

@@ -46,8 +46,26 @@ cd \$SCRIPT_PATH
 exec ./code/CLI.py "\$@"
 EOF
 
+# Create the test script
+cat >./run_tests  <<EOF
+#!/bin/sh
+# Find the executable
+if [ $(uname) = 'Darwin' ]; then
+    alias readlink='greadlink'
+fi
+ID2T_DIR=\$(readlink -f \$0)
+SCRIPT_PATH=\${ID2T_DIR%/*}
+cd \$SCRIPT_PATH/code
+# Execute tests
+set -e
+PYTHONWARNINGS="ignore" coverage run --source=. -m unittest discover -s Test/ >/dev/null
+coverage html
+coverage report -m
+EOF
+
 chmod +x ./code/CLI.py
 chmod +x ./id2t
+chmod +x ./run_tests
 
 echo -e "\n\nAll is set. ID2T is ready."
 echo -e "\nRun ID2T with the command './id2t'"

code/.coveragerc

@@ -0,0 +1,6 @@
+[run]
+disable_warnings = ResourceWarning
+
+[html]
+directory = ../htmlcov
+

code/Attack/BaseAttack.py

@@ -14,6 +14,8 @@ from scapy.utils import PcapWriter
 from Attack import AttackParameters
 from Attack.AttackParameters import Parameter
 from Attack.AttackParameters import ParameterTypes
+from ID2TLib.Utility import handle_most_used_outputs
+from lea import Lea
 import ID2TLib.libpcapreader as pr
 
 
@@ -142,10 +144,15 @@ class BaseAttack(metaclass=ABCMeta):
             """
             return num < 1 or num > 65535
 
+        if ports_input is None or ports_input is "":
+            return False
+
         if isinstance(ports_input, str):
             ports_input = ports_input.replace(' ', '').split(',')
         elif isinstance(ports_input, int):
             ports_input = [ports_input]
+        elif len(ports_input) is 0:
+            return False
 
         ports_output = []
 
@@ -246,6 +253,13 @@ class BaseAttack(metaclass=ABCMeta):
     # HELPER METHODS
     #########################################
 
+    def set_seed(self, seed: int):
+        """
+        :param seed: The random seed to be set.
+        """
+        if isinstance(seed, int):
+            random.seed(seed)
+
     def add_param_value(self, param, value):
         """
         Adds the pair param : value to the dictionary of attack parameters. Prints and error message and skips the
@@ -475,7 +489,7 @@ class BaseAttack(metaclass=ABCMeta):
             if ip_source == ip_destination:
                 equal = True
         if equal:
-            print("\nERROR: Invalid IP addresses; source IP is the same as destination IP: " + ip_source + ".")
+            print("\nERROR: Invalid IP addresses; source IP is the same as destination IP: " + ip_destination + ".")
             sys.exit(0)
 
 
@@ -542,6 +556,37 @@ class BaseAttack(metaclass=ABCMeta):
             str_tcp_seg = self.clean_white_spaces(str_tcp_seg)
         return str_tcp_seg
 
+    def get_ip_data(self, ip_address: str):
+        """
+        :param ip_address: the ip of which (packet-)data shall be returned
+        :return: MSS, TTL and Window Size values of the given IP
+        """
+        # Set MSS (Maximum Segment Size) based on MSS distribution of IP address
+        mss_dist = self.statistics.get_mss_distribution(ip_address)
+        if len(mss_dist) > 0:
+            mss_prob_dict = Lea.fromValFreqsDict(mss_dist)
+            mss_value = mss_prob_dict.random()
+        else:
+            mss_value = handle_most_used_outputs(self.statistics.process_db_query("most_used(mssValue)"))
+
+        # Set TTL based on TTL distribution of IP address
+        ttl_dist = self.statistics.get_ttl_distribution(ip_address)
+        if len(ttl_dist) > 0:
+            ttl_prob_dict = Lea.fromValFreqsDict(ttl_dist)
+            ttl_value = ttl_prob_dict.random()
+        else:
+            ttl_value = handle_most_used_outputs(self.statistics.process_db_query("most_used(ttlValue)"))
+
+        # Set Window Size based on Window Size distribution of IP address
+        win_dist = self.statistics.get_win_distribution(ip_address)
+        if len(win_dist) > 0:
+            win_prob_dict = Lea.fromValFreqsDict(win_dist)
+            win_value = win_prob_dict.random()
+        else:
+            win_value = handle_most_used_outputs(self.statistics.process_db_query("most_used(winSize)"))
+
+        return mss_value, ttl_value, win_value
+
 
     #########################################
     # RANDOM IP/MAC ADDRESS GENERATORS

code/Attack/DDoSAttack.py

@@ -9,7 +9,9 @@ from scapy.layers.inet import IP, Ether, TCP, RandShort
 from Attack import BaseAttack
 from Attack.AttackParameters import Parameter as Param
 from Attack.AttackParameters import ParameterTypes
-from ID2TLib.Utility import update_timestamp, get_interval_pps, get_nth_random_element, index_increment
+from ID2TLib.Utility import update_timestamp, get_interval_pps, get_nth_random_element, index_increment, \
+    handle_most_used_outputs, get_attacker_config
+from ID2TLib.Utility import update_timestamp, get_interval_pps, get_nth_random_element
 
 logging.getLogger("scapy.runtime").setLevel(logging.ERROR)
 # noinspection PyPep8
@@ -26,7 +28,7 @@ class DDoSAttack(BaseAttack.BaseAttack):
                                         "Resource Exhaustion")
 
         # Define allowed parameters and their type
-        self.supported_params = {
+        self.supported_params.update({
             Param.IP_SOURCE: ParameterTypes.TYPE_IP_ADDRESS,
             Param.MAC_SOURCE: ParameterTypes.TYPE_MAC_ADDRESS,
             Param.PORT_SOURCE: ParameterTypes.TYPE_PORT,
@@ -39,7 +41,7 @@ class DDoSAttack(BaseAttack.BaseAttack):
             Param.NUMBER_ATTACKERS: ParameterTypes.TYPE_INTEGER_POSITIVE,
             Param.ATTACK_DURATION: ParameterTypes.TYPE_INTEGER_POSITIVE,
             Param.VICTIM_BUFFER: ParameterTypes.TYPE_INTEGER_POSITIVE
-        }
+        })
 
     def init_params(self):
         """
@@ -55,7 +57,7 @@ class DDoSAttack(BaseAttack.BaseAttack):
         # attacker configuration
         num_attackers = randint(1, 16)
         # The most used IP class in background traffic
-        most_used_ip_class = self.statistics.process_db_query("most_used(ipClass)")
+        most_used_ip_class = handle_most_used_outputs(self.statistics.process_db_query("most_used(ipClass)"))
 
         self.add_param_value(Param.IP_SOURCE, self.generate_random_ipv4_address(most_used_ip_class, num_attackers))
         self.add_param_value(Param.MAC_SOURCE, self.generate_random_mac_address(num_attackers))
@@ -73,44 +75,13 @@ class DDoSAttack(BaseAttack.BaseAttack):
         self.add_param_value(Param.VICTIM_BUFFER, randint(1000,10000))
 
     def generate_attack_pcap(self):
-        def get_attacker_config(ipAddress: str):
-            """
-            Returns the attacker configuration depending on the IP address, this includes the port for the next
-            attacking packet and the previously used (fixed) TTL value.
-            :param ipAddress: The IP address of the attacker
-            :return: A tuple consisting of (port, ttlValue)
-            """
-            # Determine port
-            port = attacker_port_mapping.get(ipAddress)
-            if port is not None:  # use next port
-                next_port = attacker_port_mapping.get(ipAddress) + 1
-                if next_port > (2 ** 16 - 1):
-                    next_port = 1
-            else:  # generate starting port
-                next_port = RandShort()
-            attacker_port_mapping[ipAddress] = next_port
-            # Determine TTL value
-            ttl = attacker_ttl_mapping.get(ipAddress)
-            if ttl is None:  # determine TTL value
-                is_invalid = True
-                pos = ip_source_list.index(ipAddress)
-                pos_max = len(gd)
-                while is_invalid:
-                    ttl = int(round(gd[pos]))
-                    if 0 < ttl < 256:  # validity check
-                        is_invalid = False
-                    else:
-                        pos = index_increment(pos, pos_max)
-                attacker_ttl_mapping[ipAddress] = ttl
-            # return port and TTL
-            return next_port, ttl
         BUFFER_SIZE = 1000
 
         # Determine source IP and MAC address
         num_attackers = self.get_param_value(Param.NUMBER_ATTACKERS)
         if num_attackers is not None:  # user supplied Param.NUMBER_ATTACKERS
             # The most used IP class in background traffic
-            most_used_ip_class = self.statistics.process_db_query("most_used(ipClass)")
+            most_used_ip_class = handle_most_used_outputs(self.statistics.process_db_query("most_used(ipClass)"))
             # Create random attackers based on user input Param.NUMBER_ATTACKERS
             ip_source_list = self.generate_random_ipv4_address(most_used_ip_class, num_attackers)
             mac_source_list = self.generate_random_mac_address(num_attackers)
@@ -147,13 +118,15 @@ class DDoSAttack(BaseAttack.BaseAttack):
         port_destination = self.get_param_value(Param.PORT_DESTINATION)
         if not port_destination:  # user did not define port_dest
             port_destination = self.statistics.process_db_query(
-                "SELECT portNumber FROM ip_ports WHERE portDirection='in' AND ipAddress='" + ip_destination + "' ORDER BY portCount DESC LIMIT 1;")
+                "SELECT portNumber FROM ip_ports WHERE portDirection='in' AND ipAddress='" + ip_destination + "' AND portCount==(SELECT MAX(portCount) FROM ip_ports WHERE portDirection='in' AND ipAddress='" + ip_destination + "');")
         if not port_destination:  # no port was retrieved
             port_destination = self.statistics.process_db_query(
-                "SELECT portNumber FROM ip_ports WHERE portDirection='in' GROUP BY portNumber ORDER BY SUM(portCount) DESC LIMIT 1;")
+                "SELECT portNumber FROM (SELECT portNumber, SUM(portCount) as occ FROM ip_ports WHERE portDirection='in' GROUP BY portNumber ORDER BY occ DESC) WHERE occ=(SELECT SUM(portCount) FROM ip_ports WHERE portDirection='in' GROUP BY portNumber ORDER BY SUM(portCount) DESC LIMIT 1);")
         if not port_destination:
             port_destination = max(1, str(RandShort()))
 
+        port_destination = handle_most_used_outputs(port_destination)
+
         attacker_port_mapping = {}
         attacker_ttl_mapping = {}
 
@@ -170,8 +143,7 @@ class DDoSAttack(BaseAttack.BaseAttack):
         attack_duration = self.get_param_value(Param.ATTACK_DURATION)
         pkts_num = int(pps * attack_duration)
 
-        source_win_sizes = self.statistics.process_db_query(
-                "SELECT DISTINCT winSize FROM tcp_win ORDER BY RANDOM() LIMIT "+str(pkts_num)+";")
+        source_win_sizes = self.statistics.get_rnd_win_size(pkts_num)
 
         destination_win_dist = self.statistics.get_win_distribution(ip_destination)
         if len(destination_win_dist) > 0:
@@ -180,11 +152,15 @@ class DDoSAttack(BaseAttack.BaseAttack):
         else:
             destination_win_value = self.statistics.process_db_query("most_used(winSize)")
 
+        destination_win_value = handle_most_used_outputs(destination_win_value)
+
         # MSS that was used by IP destination in background traffic
         mss_dst = self.statistics.get_most_used_mss(ip_destination)
         if mss_dst is None:
             mss_dst = self.statistics.process_db_query("most_used(mssValue)")
 
+        mss_dst = handle_most_used_outputs(mss_dst)
+
         replies_count = 0
         total_pkt_num = 0
         # For each attacker, generate his own packets, then merge all packets
@@ -202,7 +178,7 @@ class DDoSAttack(BaseAttack.BaseAttack):
                 # Select one IP address and its corresponding MAC address
                 (ip_source, mac_source) = get_nth_random_element(ip_source_list, mac_source_list)
                 # Determine source port
-                (port_source, ttl_value) = get_attacker_config(ip_source)
+                (port_source, ttl_value) = get_attacker_config(ip_source_list ,ip_source)
                 request_ether = Ether(dst=mac_destination, src=mac_source)
                 request_ip = IP(src=ip_source, dst=ip_destination, ttl=ttl_value)
                 # Random win size for each packet

code/Attack/EternalBlueExploit.py

@@ -1,23 +1,23 @@
 import logging
-
 from random import randint, uniform
+
 from lea import Lea
-from scapy.utils import RawPcapReader
 from scapy.layers.inet import Ether
+from scapy.utils import RawPcapReader
 
 from Attack import BaseAttack
 from Attack.AttackParameters import Parameter as Param
 from Attack.AttackParameters import ParameterTypes
-from ID2TLib.Utility import update_timestamp, get_interval_pps
 from ID2TLib.SMBLib import smb_port
+import ID2TLib.Utility as Util
 
 logging.getLogger("scapy.runtime").setLevel(logging.ERROR)
 # noinspection PyPep8
 
 
 class EternalBlueExploit(BaseAttack.BaseAttack):
-    template_scan_pcap_path = "resources/Win7_eternalblue_scan.pcap"
-    template_attack_pcap_path = "resources/Win7_eternalblue_exploit.pcap"
+    template_scan_pcap_path = Util.RESOURCE_DIR + "Win7_eternalblue_scan.pcap"
+    template_attack_pcap_path = Util.RESOURCE_DIR + "Win7_eternalblue_exploit.pcap"
     # Empirical values from Metasploit experiments
     minDefaultPort = 30000
     maxDefaultPort = 50000
@@ -33,7 +33,7 @@ class EternalBlueExploit(BaseAttack.BaseAttack):
                                         "Privilege elevation")
 
         # Define allowed parameters and their type
-        self.supported_params = {
+        self.supported_params.update({
             Param.MAC_SOURCE: ParameterTypes.TYPE_MAC_ADDRESS,
             Param.IP_SOURCE: ParameterTypes.TYPE_IP_ADDRESS,
             Param.PORT_SOURCE: ParameterTypes.TYPE_PORT,
@@ -43,7 +43,7 @@ class EternalBlueExploit(BaseAttack.BaseAttack):
             Param.INJECT_AT_TIMESTAMP: ParameterTypes.TYPE_FLOAT,
             Param.INJECT_AFTER_PACKET: ParameterTypes.TYPE_PACKET_POSITION,
             Param.PACKETS_PER_SECOND: ParameterTypes.TYPE_FLOAT
-        }
+        })
 
     def init_params(self):
         """
@@ -57,9 +57,9 @@ class EternalBlueExploit(BaseAttack.BaseAttack):
         # (values are overwritten if user specifies them)
         # Attacker configuration
         most_used_ip_address = self.statistics.get_most_used_ip_address()
-        if isinstance(most_used_ip_address, list):
-            most_used_ip_address = most_used_ip_address[0]
         random_ip_address = self.statistics.get_random_ip_address()
+        while random_ip_address == most_used_ip_address:
+            random_ip_address = self.statistics.get_random_ip_address()
         self.add_param_value(Param.IP_SOURCE, random_ip_address)
         self.add_param_value(Param.MAC_SOURCE, self.statistics.get_mac_address(random_ip_address))
         self.add_param_value(Param.PORT_SOURCE, randint(self.minDefaultPort, self.maxDefaultPort))
@@ -108,14 +108,14 @@ class EternalBlueExploit(BaseAttack.BaseAttack):
             source_ttl_prob_dict = Lea.fromValFreqsDict(source_ttl_dist)
             source_ttl_value = source_ttl_prob_dict.random()
         else:
-            source_ttl_value = self.statistics.process_db_query("most_used(ttlValue)")
+            source_ttl_value = Util.handle_most_used_outputs(self.statistics.process_db_query("most_used(ttlValue)"))
 
         destination_ttl_dist = self.statistics.get_ttl_distribution(ip_destination)
         if len(destination_ttl_dist) > 0:
             destination_ttl_prob_dict = Lea.fromValFreqsDict(destination_ttl_dist)
             destination_ttl_value = destination_ttl_prob_dict.random()
         else:
-            destination_ttl_value = self.statistics.process_db_query("most_used(ttlValue)")
+            destination_ttl_value = Util.handle_most_used_outputs(self.statistics.process_db_query("most_used(ttlValue)"))
 
         # Set Window Size based on Window Size distribution of IP address
         source_win_dist = self.statistics.get_win_distribution(ip_source)
@@ -133,7 +133,7 @@ class EternalBlueExploit(BaseAttack.BaseAttack):
             destination_win_prob_dict = Lea.fromValFreqsDict(destination_win_dist)
 
         # Set MSS (Maximum Segment Size) based on MSS distribution of IP address
-        mss_value = self.statistics.process_db_query("most_used(mssValue)")
+        mss_value = Util.handle_most_used_outputs(self.statistics.process_db_query("most_used(mssValue)"))
         if not mss_value:
             mss_value = 1465
 
@@ -142,6 +142,7 @@ class EternalBlueExploit(BaseAttack.BaseAttack):
         orig_ip_dst = None
         exploit_raw_packets = RawPcapReader(self.template_scan_pcap_path)
         inter_arrival_times = self.get_inter_arrival_time(exploit_raw_packets)
+        exploit_raw_packets.close()
         exploit_raw_packets = RawPcapReader(self.template_scan_pcap_path)
 
         source_origin_wins, destination_origin_wins = {}, {}
@@ -183,8 +184,8 @@ class EternalBlueExploit(BaseAttack.BaseAttack):
                 new_pkt = (eth_frame / ip_pkt / tcp_pkt)
                 new_pkt.time = timestamp_next_pkt
 
-                pps = max(get_interval_pps(complement_interval_pps, timestamp_next_pkt), 10)
-                timestamp_next_pkt = update_timestamp(timestamp_next_pkt, pps) + inter_arrival_times[pkt_num]#float(timeSteps.random())
+                pps = max(Util.get_interval_pps(complement_interval_pps, timestamp_next_pkt), 10)
+                timestamp_next_pkt = Util.update_timestamp(timestamp_next_pkt, pps) + inter_arrival_times[pkt_num]#float(timeSteps.random())
             # Reply
             else:
                 # Ether
@@ -211,7 +212,7 @@ class EternalBlueExploit(BaseAttack.BaseAttack):
                         tcp_pkt.setfieldval("options", tcp_options)
 
                 new_pkt = (eth_frame / ip_pkt / tcp_pkt)
-                timestamp_next_pkt = update_timestamp(timestamp_next_pkt, pps) + inter_arrival_times[pkt_num]#+ float(timeSteps.random())
+                timestamp_next_pkt = Util.update_timestamp(timestamp_next_pkt, pps) + inter_arrival_times[pkt_num]#+ float(timeSteps.random())
                 new_pkt.time = timestamp_next_pkt
 
             packets.append(new_pkt)
@@ -219,11 +220,13 @@ class EternalBlueExploit(BaseAttack.BaseAttack):
 
         # Inject EternalBlue exploit packets
         # Read Win7_eternalblue_exploit pcap file
+        exploit_raw_packets.close()
         exploit_raw_packets = RawPcapReader(self.template_attack_pcap_path)
 
         port_source = randint(self.minDefaultPort,self.maxDefaultPort) # experiments show this range of ports
         # conversations = {(ip.src, ip.dst, port.src, port.dst): packets}
         conversations, orderList_conversations = self.packetsToConvs(exploit_raw_packets)
+        exploit_raw_packets.close()
 
         conv_start_timesamp = timestamp_next_pkt
         for conv_index, conv in enumerate(orderList_conversations):
@@ -275,8 +278,8 @@ class EternalBlueExploit(BaseAttack.BaseAttack):
                         new_pkt = (eth_frame / ip_pkt / tcp_pkt)
                         new_pkt.time = timestamp_next_pkt
 
-                        pps = max(get_interval_pps(complement_interval_pps, timestamp_next_pkt), 10)
-                        timestamp_next_pkt = update_timestamp(timestamp_next_pkt, pps) + inter_arrival_times[pkt_num] #float(timeSteps.random())
+                        pps = max(Util.get_interval_pps(complement_interval_pps, timestamp_next_pkt), 10)
+                        timestamp_next_pkt = Util.update_timestamp(timestamp_next_pkt, pps) + inter_arrival_times[pkt_num] #float(timeSteps.random())
 
                     # Reply
                     else:
@@ -305,8 +308,8 @@ class EternalBlueExploit(BaseAttack.BaseAttack):
 
                         new_pkt = (eth_frame / ip_pkt / tcp_pkt)
 
-                        pps = max(get_interval_pps(complement_interval_pps, timestamp_next_pkt), 10)
-                        timestamp_next_pkt = update_timestamp(timestamp_next_pkt, pps) + inter_arrival_times[pkt_num]#float(timeSteps.random())
+                        pps = max(Util.get_interval_pps(complement_interval_pps, timestamp_next_pkt), 10)
+                        timestamp_next_pkt = Util.update_timestamp(timestamp_next_pkt, pps) + inter_arrival_times[pkt_num]#float(timeSteps.random())
 
                         new_pkt.time = timestamp_next_pkt
 
@@ -348,8 +351,8 @@ class EternalBlueExploit(BaseAttack.BaseAttack):
                         new_pkt = (eth_frame / ip_pkt / tcp_pkt)
                         new_pkt.time = timestamp_next_pkt
 
-                        pps = max(get_interval_pps(complement_interval_pps, timestamp_next_pkt), 10)
-                        timestamp_next_pkt = update_timestamp(timestamp_next_pkt, pps) + inter_arrival_times[pkt_num]# float(timeSteps.random())
+                        pps = max(Util.get_interval_pps(complement_interval_pps, timestamp_next_pkt), 10)
+                        timestamp_next_pkt = Util.update_timestamp(timestamp_next_pkt, pps) + inter_arrival_times[pkt_num]# float(timeSteps.random())
 
                     # Reply
                     else:
@@ -378,8 +381,8 @@ class EternalBlueExploit(BaseAttack.BaseAttack):
 
                         new_pkt = (eth_frame / ip_pkt / tcp_pkt)
 
-                        pps = max(get_interval_pps(complement_interval_pps, timestamp_next_pkt), 10)
-                        timestamp_next_pkt = update_timestamp(timestamp_next_pkt, pps) + inter_arrival_times[pkt_num]# float(timeSteps.random())
+                        pps = max(Util.get_interval_pps(complement_interval_pps, timestamp_next_pkt), 10)
+                        timestamp_next_pkt = Util.update_timestamp(timestamp_next_pkt, pps) + inter_arrival_times[pkt_num]# float(timeSteps.random())
 
                         new_pkt.time = timestamp_next_pkt
 

code/Attack/FTPWinaXeExploit.py

@@ -1,4 +1,5 @@
 import logging
+import ID2TLib.Utility
 
 from random import randint
 from lea import Lea
@@ -8,7 +9,7 @@ from Attack import BaseAttack
 from Attack.AttackParameters import Parameter as Param
 from Attack.AttackParameters import ParameterTypes
 from ID2TLib.Utility import update_timestamp, generate_source_port_from_platform, get_rnd_x86_nop, forbidden_chars,\
-    get_rnd_bytes, get_bytes_from_file
+    get_rnd_bytes , check_payload_len, handle_most_used_outputs
 
 logging.getLogger("scapy.runtime").setLevel(logging.ERROR)
 # noinspection PyPep8
@@ -28,7 +29,7 @@ class FTPWinaXeExploit(BaseAttack.BaseAttack):
                                                "Privilege elevation")
 
         # Define allowed parameters and their type
-        self.supported_params = {
+        self.supported_params.update({
             Param.IP_SOURCE: ParameterTypes.TYPE_IP_ADDRESS,
             Param.IP_DESTINATION: ParameterTypes.TYPE_IP_ADDRESS,
             Param.MAC_SOURCE: ParameterTypes.TYPE_MAC_ADDRESS,
@@ -39,7 +40,7 @@ class FTPWinaXeExploit(BaseAttack.BaseAttack):
             Param.PACKETS_PER_SECOND: ParameterTypes.TYPE_FLOAT,
             Param.CUSTOM_PAYLOAD: ParameterTypes.TYPE_STRING,
             Param.CUSTOM_PAYLOAD_FILE: ParameterTypes.TYPE_STRING
-        }
+        })
 
     def init_params(self):
         """
@@ -51,11 +52,9 @@ class FTPWinaXeExploit(BaseAttack.BaseAttack):
         # PARAMETERS: initialize with default values
         # (values are overwritten if user specifies them)
         most_used_ip_address = self.statistics.get_most_used_ip_address()
-        if isinstance(most_used_ip_address, list):
-            most_used_ip_address = most_used_ip_address[0]
 
         # The most used IP class in background traffic
-        most_used_ip_class = self.statistics.process_db_query("most_used(ipClass)")
+        most_used_ip_class = handle_most_used_outputs(self.statistics.process_db_query("most_used(ipClass)"))
         attacker_ip = self.generate_random_ipv4_address(most_used_ip_class)
         self.add_param_value(Param.IP_DESTINATION, attacker_ip)
         self.add_param_value(Param.MAC_DESTINATION, self.generate_random_mac_address())
@@ -79,48 +78,6 @@ class FTPWinaXeExploit(BaseAttack.BaseAttack):
         self.add_param_value(Param.CUSTOM_PAYLOAD_FILE, '')
 
     def generate_attack_pcap(self):
-        def get_ip_data(ip_address: str):
-            """
-            :param ip_address: The ip of which (packet-)data shall be returned
-            :return: MSS, TTL and window size values of the given IP
-            """
-            # Set MSS (Maximum Segment Size) based on MSS distribution of IP address
-            mss_dist = self.statistics.get_mss_distribution(ip_address)
-            if len(mss_dist) > 0:
-                mss_prob_dict = Lea.fromValFreqsDict(mss_dist)
-                mss_value = mss_prob_dict.random()
-            else:
-                mss_value = self.statistics.process_db_query("most_used(mssValue)")
-
-            # Set TTL based on TTL distribution of IP address
-            ttl_dist = self.statistics.get_ttl_distribution(ip_address)
-            if len(ttl_dist) > 0:
-                ttl_prob_dict = Lea.fromValFreqsDict(ttl_dist)
-                ttl_value = ttl_prob_dict.random()
-            else:
-                ttl_value = self.statistics.process_db_query("most_used(ttlValue)")
-
-            # Set Window Size based on Window Size distribution of IP address
-            win_dist = self.statistics.get_win_distribution(ip_address)
-            if len(win_dist) > 0:
-                win_prob_dict = Lea.fromValFreqsDict(win_dist)
-                win_value = win_prob_dict.random()
-            else:
-                win_value = self.statistics.process_db_query("most_used(winSize)")
-
-            return mss_value, ttl_value, win_value
-
-        def check_payload_len(payload_len: int, limit: int):
-            """
-            Checks if the len of the payload exceeds a given limit
-            :param payload_len: The length of the payload
-            :param limit: The limit of the length of the payload which is allowed
-            """
-
-            if payload_len > limit:
-                print("\nCustom payload too long: ", payload_len, " bytes. Should be a maximum of ", limit, " bytes.")
-                exit(1)
-
 
         pps = self.get_param_value(Param.PACKETS_PER_SECOND)
 
@@ -145,13 +102,13 @@ class FTPWinaXeExploit(BaseAttack.BaseAttack):
         # Create random victim if specified
         if self.get_param_value(Param.IP_SOURCE_RANDOMIZE):
             # The most used IP class in background traffic
-            most_used_ip_class = self.statistics.process_db_query("most_used(ipClass)")
+            most_used_ip_class = handle_most_used_outputs(self.statistics.process_db_query("most_used(ipClass)"))
             ip_victim = self.generate_random_ipv4_address(most_used_ip_class, 1)
             mac_victim = self.generate_random_mac_address()
 
         # Get MSS, TTL and Window size value for victim/attacker IP
-        victim_mss_value, victim_ttl_value, victim_win_value = get_ip_data(ip_victim)
-        attacker_mss_value, attacker_ttl_value, attacker_win_value = get_ip_data(ip_attacker)
+        victim_mss_value, victim_ttl_value, victim_win_value = self.get_ip_data(ip_victim)
+        attacker_mss_value, attacker_ttl_value, attacker_win_value = self.get_ip_data(ip_attacker)
 
         minDelay, maxDelay = self.get_reply_delay(ip_attacker)
 
@@ -205,7 +162,7 @@ class FTPWinaXeExploit(BaseAttack.BaseAttack):
             if custom_payload_file == '':
                 payload = get_rnd_bytes(custom_payload_limit, forbidden_chars)
             else:
-                payload = get_bytes_from_file(custom_payload_file)
+                payload = ID2TLib.Utility.get_bytes_from_file(custom_payload_file)
                 check_payload_len(len(payload), custom_payload_limit)
                 payload += get_rnd_x86_nop(custom_payload_limit - len(payload), False, forbidden_chars)
         else:

code/Attack/JoomlaRegPrivExploit.py

@@ -1,21 +1,21 @@
 import logging
-
 from random import randint
+
 from lea import Lea
-from scapy.utils import RawPcapReader
 from scapy.layers.inet import Ether
+from scapy.utils import RawPcapReader
 
 from Attack import BaseAttack
 from Attack.AttackParameters import Parameter as Param
 from Attack.AttackParameters import ParameterTypes
-from ID2TLib.Utility import update_timestamp, get_interval_pps
+import ID2TLib.Utility as Util
 
 logging.getLogger("scapy.runtime").setLevel(logging.ERROR)
 # noinspection PyPep8
 
 
 class JoomlaRegPrivExploit(BaseAttack.BaseAttack):
-    template_attack_pcap_path = "resources/joomla_registration_privesc.pcap"
+    template_attack_pcap_path = Util.RESOURCE_DIR + "joomla_registration_privesc.pcap"
     # HTTP port
     http_port = 80
     # Metasploit experiments show this range of ports
@@ -32,7 +32,7 @@ class JoomlaRegPrivExploit(BaseAttack.BaseAttack):
                                         "Privilege elevation")
 
         # Define allowed parameters and their type
-        self.supported_params = {
+        self.supported_params.update({
             Param.MAC_SOURCE: ParameterTypes.TYPE_MAC_ADDRESS,
             Param.IP_SOURCE: ParameterTypes.TYPE_IP_ADDRESS,
             Param.MAC_DESTINATION: ParameterTypes.TYPE_MAC_ADDRESS,
@@ -43,7 +43,7 @@ class JoomlaRegPrivExploit(BaseAttack.BaseAttack):
             Param.INJECT_AT_TIMESTAMP: ParameterTypes.TYPE_FLOAT,
             Param.INJECT_AFTER_PACKET: ParameterTypes.TYPE_PACKET_POSITION,
             Param.PACKETS_PER_SECOND: ParameterTypes.TYPE_FLOAT
-        }
+        })
 
     def init_params(self):
         """
@@ -57,8 +57,6 @@ class JoomlaRegPrivExploit(BaseAttack.BaseAttack):
         # (values are overwritten if user specifies them)
         # Attacker configuration
         most_used_ip_address = self.statistics.get_most_used_ip_address()
-        if isinstance(most_used_ip_address, list):
-            most_used_ip_address = most_used_ip_address[0]
         self.add_param_value(Param.IP_SOURCE, most_used_ip_address)
         self.add_param_value(Param.MAC_SOURCE, self.statistics.get_mac_address(most_used_ip_address))
 
@@ -112,20 +110,21 @@ class JoomlaRegPrivExploit(BaseAttack.BaseAttack):
             source_ttl_prob_dict = Lea.fromValFreqsDict(source_ttl_dist)
             source_ttl_value = source_ttl_prob_dict.random()
         else:
-            source_ttl_value = self.statistics.process_db_query("most_used(ttlValue)")
+            source_ttl_value = Util.handle_most_used_outputs(self.statistics.process_db_query("most_used(ttlValue)"))
 
         destination_ttl_dist = self.statistics.get_ttl_distribution(ip_destination)
         if len(destination_ttl_dist) > 0:
             destination_ttl_prob_dict = Lea.fromValFreqsDict(destination_ttl_dist)
             destination_ttl_value = destination_ttl_prob_dict.random()
         else:
-            destination_ttl_value = self.statistics.process_db_query("most_used(ttlValue)")
+            destination_ttl_value = Util.handle_most_used_outputs(self.statistics.process_db_query("most_used(ttlValue)"))
 
         # Inject Joomla_registration_privesc
         # Read joomla_registration_privesc pcap file
         orig_ip_dst = None
         exploit_raw_packets = RawPcapReader(self.template_attack_pcap_path)
         inter_arrival_times, inter_arrival_time_dist = self.get_inter_arrival_time(exploit_raw_packets,True)
+        exploit_raw_packets.close()
         timeSteps = Lea.fromValFreqsDict(inter_arrival_time_dist)
         exploit_raw_packets = RawPcapReader(self.template_attack_pcap_path)
 
@@ -186,8 +185,8 @@ class JoomlaRegPrivExploit(BaseAttack.BaseAttack):
                 new_pkt = (eth_frame / ip_pkt/ tcp_pkt / str_tcp_seg)
                 new_pkt.time = timestamp_next_pkt
 
-                pps = max(get_interval_pps(complement_interval_pps, timestamp_next_pkt), 10)
-                timestamp_next_pkt = update_timestamp(timestamp_next_pkt, pps) + float(timeSteps.random())
+                pps = max(Util.get_interval_pps(complement_interval_pps, timestamp_next_pkt), 10)
+                timestamp_next_pkt = Util.update_timestamp(timestamp_next_pkt, pps) + float(timeSteps.random())
 
             # Reply: Victim --> attacker
             else:
@@ -212,12 +211,13 @@ class JoomlaRegPrivExploit(BaseAttack.BaseAttack):
                     victim_seq += max(strLen, 1)
 
                 new_pkt = (eth_frame / ip_pkt / tcp_pkt / str_tcp_seg)
-                pps = max(get_interval_pps(complement_interval_pps, timestamp_next_pkt), 10)
-                timestamp_next_pkt = update_timestamp(timestamp_next_pkt, pps) + float(timeSteps.random())
+                pps = max(Util.get_interval_pps(complement_interval_pps, timestamp_next_pkt), 10)
+                timestamp_next_pkt = Util.update_timestamp(timestamp_next_pkt, pps) + float(timeSteps.random())
                 new_pkt.time = timestamp_next_pkt
 
             packets.append(new_pkt)
 
+        exploit_raw_packets.close()
         # Store timestamp of first packet (for attack label)
         self.attack_start_utime = packets[0].time
         self.attack_end_utime = packets[-1].time

code/Attack/PortscanAttack.py

@@ -1,14 +1,14 @@
-import logging
 import csv
-
+import logging
 from random import shuffle, randint, choice
+
 from lea import Lea
 from scapy.layers.inet import IP, Ether, TCP
 
 from Attack import BaseAttack
 from Attack.AttackParameters import Parameter as Param
 from Attack.AttackParameters import ParameterTypes
-from ID2TLib.Utility import update_timestamp, get_interval_pps
+import ID2TLib.Utility as Util
 
 logging.getLogger("scapy.runtime").setLevel(logging.ERROR)
 # noinspection PyPep8
@@ -26,7 +26,7 @@ class PortscanAttack(BaseAttack.BaseAttack):
                                              "Scanning/Probing")
 
         # Define allowed parameters and their type
-        self.supported_params = {
+        self.supported_params.update({
             Param.IP_SOURCE: ParameterTypes.TYPE_IP_ADDRESS,
             Param.IP_DESTINATION: ParameterTypes.TYPE_IP_ADDRESS,
             Param.PORT_SOURCE: ParameterTypes.TYPE_PORT,
@@ -41,7 +41,7 @@ class PortscanAttack(BaseAttack.BaseAttack):
             Param.IP_SOURCE_RANDOMIZE: ParameterTypes.TYPE_BOOLEAN,
             Param.PACKETS_PER_SECOND: ParameterTypes.TYPE_FLOAT,
             Param.PORT_SOURCE_RANDOMIZE: ParameterTypes.TYPE_BOOLEAN
-        }
+        })
 
     def init_params(self):
         """
@@ -54,8 +54,6 @@ class PortscanAttack(BaseAttack.BaseAttack):
         # PARAMETERS: initialize with default values
         # (values are overwritten if user specifies them)
         most_used_ip_address = self.statistics.get_most_used_ip_address()
-        if isinstance(most_used_ip_address, list):
-            most_used_ip_address = most_used_ip_address[0]
 
         self.add_param_value(Param.IP_SOURCE, most_used_ip_address)
         self.add_param_value(Param.IP_SOURCE_RANDOMIZE, 'False')
@@ -89,12 +87,14 @@ class PortscanAttack(BaseAttack.BaseAttack):
         :return: Ports numbers to be used as default destination ports or default open ports in the port scan.
         """
         ports_dst = []
-        spamreader = csv.reader(open('resources/nmap-services-tcp.csv', 'rt'), delimiter=',')
+        file = open(Util.RESOURCE_DIR + 'nmap-services-tcp.csv', 'rt')
+        spamreader = csv.reader(file, delimiter=',')
         for count in range(ports_num):
             # escape first row (header)
             next(spamreader)
             # save ports numbers
             ports_dst.append(next(spamreader)[0])
+        file.close()
         # shuffle ports numbers partially
         if (ports_num == 1000):  # used for port.dst
             temp_array = [[0 for i in range(10)] for i in range(100)]
@@ -109,10 +109,6 @@ class PortscanAttack(BaseAttack.BaseAttack):
         return port_dst_shuffled
 
     def generate_attack_pcap(self):
-
-
-
-
         mac_source = self.get_param_value(Param.MAC_SOURCE)
         mac_destination = self.get_param_value(Param.MAC_DESTINATION)
         pps = self.get_param_value(Param.PACKETS_PER_SECOND)
@@ -170,13 +166,13 @@ class PortscanAttack(BaseAttack.BaseAttack):
             source_mss_prob_dict = Lea.fromValFreqsDict(source_mss_dist)
             source_mss_value = source_mss_prob_dict.random()
         else:
-            source_mss_value = self.statistics.process_db_query("most_used(mssValue)")
+            source_mss_value = Util.handle_most_used_outputs(self.statistics.process_db_query("most_used(mssValue)"))
         destination_mss_dist = self.statistics.get_mss_distribution(ip_destination)
         if len(destination_mss_dist) > 0:
             destination_mss_prob_dict = Lea.fromValFreqsDict(destination_mss_dist)
             destination_mss_value = destination_mss_prob_dict.random()
         else:
-            destination_mss_value = self.statistics.process_db_query("most_used(mssValue)")
+            destination_mss_value = Util.handle_most_used_outputs(self.statistics.process_db_query("most_used(mssValue)"))
 
         # Set TTL based on TTL distribution of IP address
         source_ttl_dist = self.statistics.get_ttl_distribution(ip_source)
@@ -184,13 +180,13 @@ class PortscanAttack(BaseAttack.BaseAttack):
             source_ttl_prob_dict = Lea.fromValFreqsDict(source_ttl_dist)
             source_ttl_value = source_ttl_prob_dict.random()
         else:
-            source_ttl_value = self.statistics.process_db_query("most_used(ttlValue)")
+            source_ttl_value = Util.handle_most_used_outputs(self.statistics.process_db_query("most_used(ttlValue)"))
         destination_ttl_dist = self.statistics.get_ttl_distribution(ip_destination)
         if len(destination_ttl_dist) > 0:
             destination_ttl_prob_dict = Lea.fromValFreqsDict(destination_ttl_dist)
             destination_ttl_value = destination_ttl_prob_dict.random()
         else:
-            destination_ttl_value = self.statistics.process_db_query("most_used(ttlValue)")
+            destination_ttl_value = Util.handle_most_used_outputs(self.statistics.process_db_query("most_used(ttlValue)"))
 
         # Set Window Size based on Window Size distribution of IP address
         source_win_dist = self.statistics.get_win_distribution(ip_source)
@@ -198,13 +194,13 @@ class PortscanAttack(BaseAttack.BaseAttack):
             source_win_prob_dict = Lea.fromValFreqsDict(source_win_dist)
             source_win_value = source_win_prob_dict.random()
         else:
-            source_win_value = self.statistics.process_db_query("most_used(winSize)")
+            source_win_value = Util.handle_most_used_outputs(self.statistics.process_db_query("most_used(winSize)"))
         destination_win_dist = self.statistics.get_win_distribution(ip_destination)
         if len(destination_win_dist) > 0:
             destination_win_prob_dict = Lea.fromValFreqsDict(destination_win_dist)
             destination_win_value = destination_win_prob_dict.random()
         else:
-            destination_win_value = self.statistics.process_db_query("most_used(winSize)")
+            destination_win_value = Util.handle_most_used_outputs(self.statistics.process_db_query("most_used(winSize)"))
 
         minDelay,maxDelay = self.get_reply_delay(ip_destination)
 
@@ -236,9 +232,9 @@ class PortscanAttack(BaseAttack.BaseAttack):
                                     options=[('MSS', destination_mss_value)])
                 reply = (reply_ether / reply_ip / reply_tcp)
 
-                timestamp_reply = update_timestamp(timestamp_next_pkt,pps,minDelay)
+                timestamp_reply = Util.update_timestamp(timestamp_next_pkt,pps,minDelay)
                 while (timestamp_reply <= timestamp_prv_reply):
-                    timestamp_reply = update_timestamp(timestamp_prv_reply,pps,minDelay)
+                    timestamp_reply = Util.update_timestamp(timestamp_prv_reply,pps,minDelay)
                 timestamp_prv_reply = timestamp_reply
 
                 reply.time = timestamp_reply
@@ -249,14 +245,14 @@ class PortscanAttack(BaseAttack.BaseAttack):
                 confirm_ip = request_ip
                 confirm_tcp = TCP(sport=sport, dport=dport, seq=1, window=0, flags='R')
                 confirm = (confirm_ether / confirm_ip / confirm_tcp)
-                timestamp_confirm = update_timestamp(timestamp_reply,pps,minDelay)
+                timestamp_confirm = Util.update_timestamp(timestamp_reply,pps,minDelay)
                 confirm.time = timestamp_confirm
                 packets.append(confirm)
 
                 # else: destination port is NOT OPEN -> no reply is sent by target
 
-            pps = max(get_interval_pps(complement_interval_pps, timestamp_next_pkt), 10)
-            timestamp_next_pkt = update_timestamp(timestamp_next_pkt, pps)
+            pps = max(Util.get_interval_pps(complement_interval_pps, timestamp_next_pkt), 10)
+            timestamp_next_pkt = Util.update_timestamp(timestamp_next_pkt, pps)
 
         # store end time of attack
         self.attack_end_utime = packets[-1].time

code/Attack/SMBLorisAttack.py

@@ -1,6 +1,6 @@
 import logging
 
-from random import randint, uniform
+import random
 from lea import Lea
 from scapy.layers.inet import IP, Ether, TCP
 from scapy.layers.netbios import NBTSession
@@ -8,7 +8,7 @@ from scapy.layers.netbios import NBTSession
 from Attack import BaseAttack
 from Attack.AttackParameters import Parameter as Param
 from Attack.AttackParameters import ParameterTypes
-from ID2TLib.Utility import update_timestamp
+from ID2TLib.Utility import update_timestamp, handle_most_used_outputs
 from ID2TLib.SMBLib import smb_port
 
 logging.getLogger("scapy.runtime").setLevel(logging.ERROR)
@@ -27,7 +27,7 @@ class SMBLorisAttack(BaseAttack.BaseAttack):
                                              "Resource Exhaustion")
 
         # Define allowed parameters and their type
-        self.supported_params = {
+        self.supported_params.update({
             Param.IP_SOURCE: ParameterTypes.TYPE_IP_ADDRESS,
             Param.IP_DESTINATION: ParameterTypes.TYPE_IP_ADDRESS,
             Param.MAC_SOURCE: ParameterTypes.TYPE_MAC_ADDRESS,
@@ -37,25 +37,22 @@ class SMBLorisAttack(BaseAttack.BaseAttack):
             Param.PACKETS_PER_SECOND: ParameterTypes.TYPE_FLOAT,
             Param.ATTACK_DURATION: ParameterTypes.TYPE_INTEGER_POSITIVE,
             Param.NUMBER_ATTACKERS: ParameterTypes.TYPE_INTEGER_POSITIVE
-        }
+        })
 
     def init_params(self):
         """
         Initialize the parameters of this attack using the user supplied command line parameters.
         Use the provided statistics to calculate default parameters and to process user
         supplied queries.
-
-        :param statistics: Reference to a statistics object.
         """
+
         # PARAMETERS: initialize with default values
         # (values are overwritten if user specifies them)
         most_used_ip_address = self.statistics.get_most_used_ip_address()
-        if isinstance(most_used_ip_address, list):
-            most_used_ip_address = most_used_ip_address[0]
 
         # The most used IP class in background traffic
-        most_used_ip_class = self.statistics.process_db_query("most_used(ipClass)")
-        num_attackers = randint(1, 16)
+        most_used_ip_class = handle_most_used_outputs(self.statistics.process_db_query("most_used(ipClass)"))
+        num_attackers = random.randint(1, 16)
         source_ip = self.generate_random_ipv4_address(most_used_ip_class, num_attackers)
 
         self.add_param_value(Param.IP_SOURCE, source_ip)
@@ -74,40 +71,10 @@ class SMBLorisAttack(BaseAttack.BaseAttack):
         self.add_param_value(Param.PACKETS_PER_SECOND,
                              (self.statistics.get_pps_sent(most_used_ip_address) +
                               self.statistics.get_pps_received(most_used_ip_address)) / 2)
-        self.add_param_value(Param.INJECT_AFTER_PACKET, randint(0, self.statistics.get_packet_count()))
+        self.add_param_value(Param.INJECT_AFTER_PACKET, random.randint(0, self.statistics.get_packet_count()))
         self.add_param_value(Param.ATTACK_DURATION, 30)
 
     def generate_attack_pcap(self):
-        def get_ip_data(ip_address: str):
-            """
-            :param ip_address: the ip of which (packet-)data shall be returned
-            :return: MSS, TTL and Window Size values of the given IP
-            """
-            # Set MSS (Maximum Segment Size) based on MSS distribution of IP address
-            mss_dist = self.statistics.get_mss_distribution(ip_address)
-            if len(mss_dist) > 0:
-                mss_prob_dict = Lea.fromValFreqsDict(mss_dist)
-                mss_value = mss_prob_dict.random()
-            else:
-                mss_value = self.statistics.process_db_query("most_used(mssValue)")
-
-            # Set TTL based on TTL distribution of IP address
-            ttl_dist = self.statistics.get_ttl_distribution(ip_address)
-            if len(ttl_dist) > 0:
-                ttl_prob_dict = Lea.fromValFreqsDict(ttl_dist)
-                ttl_value = ttl_prob_dict.random()
-            else:
-                ttl_value = self.statistics.process_db_query("most_used(ttlValue)")
-
-            # Set Window Size based on Window Size distribution of IP address
-            win_dist = self.statistics.get_win_distribution(ip_address)
-            if len(win_dist) > 0:
-                win_prob_dict = Lea.fromValFreqsDict(win_dist)
-                win_value = win_prob_dict.random()
-            else:
-                win_value = self.statistics.process_db_query("most_used(winSize)")
-
-            return mss_value, ttl_value, win_value
 
         pps = self.get_param_value(Param.PACKETS_PER_SECOND)
 
@@ -125,7 +92,7 @@ class SMBLorisAttack(BaseAttack.BaseAttack):
         num_attackers = self.get_param_value(Param.NUMBER_ATTACKERS)
         if (num_attackers is not None) and (num_attackers is not 0):  # user supplied Param.NUMBER_ATTACKERS
             # The most used IP class in background traffic
-            most_used_ip_class = self.statistics.process_db_query("most_used(ipClass)")
+            most_used_ip_class = handle_most_used_outputs(self.statistics.process_db_query("most_used(ipClass)"))
             # Create random attackers based on user input Param.NUMBER_ATTACKERS
             ip_source = self.generate_random_ipv4_address(most_used_ip_class, num_attackers)
             mac_source = self.generate_random_mac_address(num_attackers)
@@ -155,7 +122,7 @@ class SMBLorisAttack(BaseAttack.BaseAttack):
         self.ip_src_dst_equal_check(ip_source_list, ip_destination)
 
         # Get MSS, TTL and Window size value for destination IP
-        destination_mss_value, destination_ttl_value, destination_win_value = get_ip_data(ip_destination)
+        destination_mss_value, destination_ttl_value, destination_win_value = self.get_ip_data(ip_destination)
 
         minDelay,maxDelay = self.get_reply_delay(ip_destination)
 
@@ -166,15 +133,15 @@ class SMBLorisAttack(BaseAttack.BaseAttack):
 
         for attacker in range(num_attackers):
             # Get MSS, TTL and Window size value for source IP(attacker)
-            source_mss_value, source_ttl_value, source_win_value = get_ip_data(ip_source_list[attacker])
+            source_mss_value, source_ttl_value, source_win_value = self.get_ip_data(ip_source_list[attacker])
 
-            attacker_seq = randint(1000, 50000)
-            victim_seq = randint(1000, 50000)
+            attacker_seq = random.randint(1000, 50000)
+            victim_seq = random.randint(1000, 50000)
 
             sport = 1025
 
             # Timestamps of first packets shouldn't be exactly the same to look more realistic
-            timestamp_next_pkt = uniform(first_timestamp, update_timestamp(first_timestamp, pps))
+            timestamp_next_pkt = random.uniform(first_timestamp, update_timestamp(first_timestamp, pps))
 
             while timestamp_next_pkt <= attack_ends_time:
                 # Establish TCP connection

code/Attack/SMBScanAttack.py

@@ -10,8 +10,9 @@ from Attack import BaseAttack
 from Attack.AttackParameters import Parameter as Param
 from Attack.AttackParameters import ParameterTypes
 from ID2TLib.SMB2 import *
-from ID2TLib.Utility import update_timestamp, get_interval_pps, get_rnd_os, get_ip_range,\
-    generate_source_port_from_platform, get_filetime_format
+from ID2TLib.Utility import update_timestamp, get_interval_pps, get_ip_range,\
+    generate_source_port_from_platform, get_filetime_format, handle_most_used_outputs
+import ID2TLib.Utility
 from ID2TLib.SMBLib import smb_port, smb_versions, smb_dialects, get_smb_version, get_smb_platform_data,\
     invalid_smb_version
 
@@ -31,7 +32,7 @@ class SMBScanAttack(BaseAttack.BaseAttack):
                                              "Scanning/Probing")
 
         # Define allowed parameters and their type
-        self.supported_params = {
+        self.supported_params.update({
             Param.IP_SOURCE: ParameterTypes.TYPE_IP_ADDRESS,
             Param.IP_DESTINATION: ParameterTypes.TYPE_IP_ADDRESS,
             Param.PORT_SOURCE: ParameterTypes.TYPE_PORT,
@@ -47,21 +48,18 @@ class SMBScanAttack(BaseAttack.BaseAttack):
             Param.SOURCE_PLATFORM: ParameterTypes.TYPE_STRING,
             Param.PROTOCOL_VERSION: ParameterTypes.TYPE_STRING,
             Param.IP_DESTINATION_END: ParameterTypes.TYPE_IP_ADDRESS
-        }
+        })
 
     def init_params(self):
         """
         Initialize the parameters of this attack using the user supplied command line parameters.
         Use the provided statistics to calculate default parameters and to process user
         supplied queries.
-
-        :param statistics: Reference to a statistics object.
         """
+
         # PARAMETERS: initialize with default values
         # (values are overwritten if user specifies them)
         most_used_ip_address = self.statistics.get_most_used_ip_address()
-        if isinstance(most_used_ip_address, list):
-            most_used_ip_address = most_used_ip_address[0]
 
         self.add_param_value(Param.IP_SOURCE, most_used_ip_address)
         self.add_param_value(Param.IP_SOURCE_RANDOMIZE, 'False')
@@ -87,47 +85,15 @@ class SMBScanAttack(BaseAttack.BaseAttack):
                               self.statistics.get_pps_received(most_used_ip_address)) / 2)
         self.add_param_value(Param.INJECT_AFTER_PACKET, randint(0, self.statistics.get_packet_count()))
 
-        rnd_ip_count = self.statistics.get_ip_address_count()/2
+        rnd_ip_count = self.statistics.get_ip_address_count()//2
         self.add_param_value(Param.HOSTING_IP, self.statistics.get_random_ip_address(rnd_ip_count))
-        self.host_os = get_rnd_os()
+        self.host_os = ID2TLib.Utility.get_rnd_os()
         self.add_param_value(Param.HOSTING_VERSION, get_smb_version(platform=self.host_os))
-        self.add_param_value(Param.SOURCE_PLATFORM, get_rnd_os())
+        self.add_param_value(Param.SOURCE_PLATFORM, ID2TLib.Utility.get_rnd_os())
         self.add_param_value(Param.PROTOCOL_VERSION, "1")
         self.add_param_value(Param.IP_DESTINATION_END, "0.0.0.0")
 
     def generate_attack_pcap(self):
-        def get_ip_data(ip_address: str):
-            """
-            Gets the MSS, TTL and Windows Size values of a given IP
-
-            :param ip_address: the ip of which (packet-)data shall be returned
-            :return: MSS, TTL and Window Size values of the given IP
-            """
-            # Set MSS (Maximum Segment Size) based on MSS distribution of IP address
-            mss_dist = self.statistics.get_mss_distribution(ip_address)
-            if len(mss_dist) > 0:
-                mss_prob_dict = Lea.fromValFreqsDict(mss_dist)
-                mss_value = mss_prob_dict.random()
-            else:
-                mss_value = self.statistics.process_db_query("most_used(mssValue)")
-
-            # Set TTL based on TTL distribution of IP address
-            ttl_dist = self.statistics.get_ttl_distribution(ip_address)
-            if len(ttl_dist) > 0:
-                ttl_prob_dict = Lea.fromValFreqsDict(ttl_dist)
-                ttl_value = ttl_prob_dict.random()
-            else:
-                ttl_value = self.statistics.process_db_query("most_used(ttlValue)")
-
-            # Set Window Size based on Window Size distribution of IP address
-            win_dist = self.statistics.get_win_distribution(ip_address)
-            if len(win_dist) > 0:
-                win_prob_dict = Lea.fromValFreqsDict(win_dist)
-                win_value = win_prob_dict.random()
-            else:
-                win_value = self.statistics.process_db_query("most_used(winSize)")
-
-            return mss_value, ttl_value, win_value
 
         pps = self.get_param_value(Param.PACKETS_PER_SECOND)
 
@@ -195,7 +161,7 @@ class SMBScanAttack(BaseAttack.BaseAttack):
                 mac_source = self.generate_random_mac_address()
 
         # Get MSS, TTL and Window size value for source IP
-        source_mss_value, source_ttl_value, source_win_value = get_ip_data(ip_source)
+        source_mss_value, source_ttl_value, source_win_value = self.get_ip_data(ip_source)
 
         for ip in ip_dests:
 
@@ -215,7 +181,7 @@ class SMBScanAttack(BaseAttack.BaseAttack):
                         mac_destination = self.generate_random_mac_address()
 
                 # Get MSS, TTL and Window size value for destination IP
-                destination_mss_value, destination_ttl_value, destination_win_value = get_ip_data(ip)
+                destination_mss_value, destination_ttl_value, destination_win_value = self.get_ip_data(ip)
 
                 minDelay, maxDelay = self.get_reply_delay(ip)
 

code/Attack/SQLiAttack.py

@@ -1,21 +1,21 @@
 import logging
+import random
 
-from random import randint
 from lea import Lea
-from scapy.utils import RawPcapReader
 from scapy.layers.inet import Ether
-from ID2TLib.Utility import update_timestamp, get_interval_pps
+from scapy.utils import RawPcapReader
 
 from Attack import BaseAttack
 from Attack.AttackParameters import Parameter as Param
 from Attack.AttackParameters import ParameterTypes
+import ID2TLib.Utility as Util
 
 logging.getLogger("scapy.runtime").setLevel(logging.ERROR)
 # noinspection PyPep8
 
 
 class SQLiAttack(BaseAttack.BaseAttack):
-    template_attack_pcap_path = "resources/ATutorSQLi.pcap"
+    template_attack_pcap_path = Util.RESOURCE_DIR + "ATutorSQLi.pcap"
     # HTTP port
     http_port = 80
     # Metasploit experiments show this range of ports
@@ -32,7 +32,7 @@ class SQLiAttack(BaseAttack.BaseAttack):
                                         "Privilege elevation")
 
         # Define allowed parameters and their type
-        self.supported_params = {
+        self.supported_params.update({
             Param.MAC_SOURCE: ParameterTypes.TYPE_MAC_ADDRESS,
             Param.IP_SOURCE: ParameterTypes.TYPE_IP_ADDRESS,
             Param.MAC_DESTINATION: ParameterTypes.TYPE_MAC_ADDRESS,
@@ -43,22 +43,20 @@ class SQLiAttack(BaseAttack.BaseAttack):
             Param.INJECT_AT_TIMESTAMP: ParameterTypes.TYPE_FLOAT,
             Param.INJECT_AFTER_PACKET: ParameterTypes.TYPE_PACKET_POSITION,
             Param.PACKETS_PER_SECOND: ParameterTypes.TYPE_FLOAT
-        }
+        })
 
     def init_params(self):
         """
         Initialize the parameters of this attack using the user supplied command line parameters.
         Use the provided statistics to calculate default parameters and to process user
         supplied queries.
-
-        :param statistics: Reference to a statistics object.
         """
+
         # PARAMETERS: initialize with default utilsvalues
         # (values are overwritten if user specifies them)
         # Attacker configuration
         most_used_ip_address = self.statistics.get_most_used_ip_address()
-        if isinstance(most_used_ip_address, list):
-            most_used_ip_address = most_used_ip_address[0]
+
         self.add_param_value(Param.IP_SOURCE, most_used_ip_address)
         self.add_param_value(Param.MAC_SOURCE, self.statistics.get_mac_address(most_used_ip_address))
 
@@ -74,7 +72,7 @@ class SQLiAttack(BaseAttack.BaseAttack):
         self.add_param_value(Param.TARGET_HOST, "www.hackme.com")
 
         # Attack configuration
-        self.add_param_value(Param.INJECT_AFTER_PACKET, randint(0, self.statistics.get_packet_count()))
+        self.add_param_value(Param.INJECT_AFTER_PACKET, random.randint(0, self.statistics.get_packet_count()))
         self.add_param_value(Param.PACKETS_PER_SECOND,
                              (self.statistics.get_pps_sent(most_used_ip_address) +
                               self.statistics.get_pps_received(most_used_ip_address)) / 2)
@@ -109,14 +107,14 @@ class SQLiAttack(BaseAttack.BaseAttack):
             source_ttl_prob_dict = Lea.fromValFreqsDict(source_ttl_dist)
             source_ttl_value = source_ttl_prob_dict.random()
         else:
-            source_ttl_value = self.statistics.process_db_query("most_used(ttlValue)")
+            source_ttl_value = Util.handle_most_used_outputs(self.statistics.process_db_query("most_used(ttlValue)"))
 
         destination_ttl_dist = self.statistics.get_ttl_distribution(ip_destination)
         if len(destination_ttl_dist) > 0:
             destination_ttl_prob_dict = Lea.fromValFreqsDict(destination_ttl_dist)
             destination_ttl_value = destination_ttl_prob_dict.random()
         else:
-            destination_ttl_value = self.statistics.process_db_query("most_used(ttlValue)")
+            destination_ttl_value = Util.handle_most_used_outputs(self.statistics.process_db_query("most_used(ttlValue)"))
 
         # Inject SQLi Attack
         # Read SQLi Attack pcap file
@@ -124,15 +122,16 @@ class SQLiAttack(BaseAttack.BaseAttack):
         exploit_raw_packets = RawPcapReader(self.template_attack_pcap_path)
         inter_arrival_times, inter_arrival_time_dist = self.get_inter_arrival_time(exploit_raw_packets,True)
         timeSteps = Lea.fromValFreqsDict(inter_arrival_time_dist)
+        exploit_raw_packets.close()
         exploit_raw_packets = RawPcapReader(self.template_attack_pcap_path)
 
-        port_source = randint(self.minDefaultPort,self.maxDefaultPort) # experiments show this range of ports
+        port_source = random.randint(self.minDefaultPort,self.maxDefaultPort) # experiments show this range of ports
 
         # Random TCP sequence numbers
         global attacker_seq
-        attacker_seq = randint(1000, 50000)
+        attacker_seq = random.randint(1000, 50000)
         global victim_seq
-        victim_seq = randint(1000, 50000)
+        victim_seq = random.randint(1000, 50000)
 
         for pkt_num, pkt in enumerate(exploit_raw_packets):
             eth_frame = Ether(pkt[0])
@@ -156,11 +155,11 @@ class SQLiAttack(BaseAttack.BaseAttack):
 
                     # There are 363 TCP connections with different source ports, for each of them we generate random port
                     if tcp_pkt.getfieldval("sport") != prev_orig_port_source and tcp_pkt.getfieldval("dport") != 4444:
-                        port_source = randint(self.minDefaultPort, self.maxDefaultPort)
+                        port_source = random.randint(self.minDefaultPort, self.maxDefaultPort)
                         prev_orig_port_source = tcp_pkt.getfieldval("sport")
                         # New connection, new random TCP sequence numbers
-                        attacker_seq = randint(1000, 50000)
-                        victim_seq = randint(1000, 50000)
+                        attacker_seq = random.randint(1000, 50000)
+                        victim_seq = random.randint(1000, 50000)
                         # First packet in a connection has ACK = 0
                         tcp_pkt.setfieldval("ack", 0)
 
@@ -187,8 +186,8 @@ class SQLiAttack(BaseAttack.BaseAttack):
                     new_pkt = (eth_frame / ip_pkt/ tcp_pkt / str_tcp_seg)
                     new_pkt.time = timestamp_next_pkt
 
-                    pps = max(get_interval_pps(complement_interval_pps, timestamp_next_pkt), 10)
-                    timestamp_next_pkt = update_timestamp(timestamp_next_pkt, pps) + float(timeSteps.random())
+                    pps = max(Util.get_interval_pps(complement_interval_pps, timestamp_next_pkt), 10)
+                    timestamp_next_pkt = Util.update_timestamp(timestamp_next_pkt, pps) + float(timeSteps.random())
 
                 # Victim --> attacker
                 else:
@@ -213,17 +212,17 @@ class SQLiAttack(BaseAttack.BaseAttack):
                         victim_seq += max(strLen, 1)
 
                     new_pkt = (eth_frame / ip_pkt / tcp_pkt / str_tcp_seg)
-                    timestamp_next_pkt = update_timestamp(timestamp_next_pkt, pps) + float(timeSteps.random())
+                    timestamp_next_pkt = Util.update_timestamp(timestamp_next_pkt, pps) + float(timeSteps.random())
                     new_pkt.time = timestamp_next_pkt
 
             # The last connection
             else:
                 # New connection, new random TCP sequence numbers
-                attacker_seq = randint(1000, 50000)
-                victim_seq = randint(1000, 50000)
+                attacker_seq = random.randint(1000, 50000)
+                victim_seq = random.randint(1000, 50000)
                 # First packet in a connection has ACK = 0
                 tcp_pkt.setfieldval("ack", 0)
-                #port_source = randint(self.minDefaultPort, self.maxDefaultPort)
+                #port_source = random.randint(self.minDefaultPort, self.maxDefaultPort)
 
                 # Attacker --> vicitm
                 if ip_pkt.getfieldval("dst") == orig_ip_dst:  # victim IP
@@ -249,8 +248,8 @@ class SQLiAttack(BaseAttack.BaseAttack):
                     new_pkt = (eth_frame / ip_pkt / tcp_pkt / str_tcp_seg)
                     new_pkt.time = timestamp_next_pkt
 
-                    pps = max(get_interval_pps(complement_interval_pps, timestamp_next_pkt), 10)
-                    timestamp_next_pkt = update_timestamp(timestamp_next_pkt, pps) + float(timeSteps.random())
+                    pps = max(Util.get_interval_pps(complement_interval_pps, timestamp_next_pkt), 10)
+                    timestamp_next_pkt = Util.update_timestamp(timestamp_next_pkt, pps) + float(timeSteps.random())
 
                 # Victim --> attacker
                 else:
@@ -274,11 +273,12 @@ class SQLiAttack(BaseAttack.BaseAttack):
                         victim_seq += max(strLen, 1)
 
                     new_pkt = (eth_frame / ip_pkt / tcp_pkt / str_tcp_seg)
-                    timestamp_next_pkt = update_timestamp(timestamp_next_pkt, pps) + float(timeSteps.random())
+                    timestamp_next_pkt = Util.update_timestamp(timestamp_next_pkt, pps) + float(timeSteps.random())
                     new_pkt.time = timestamp_next_pkt
 
             packets.append(new_pkt)
 
+        exploit_raw_packets.close()
         # Store timestamp of first packet (for attack label)
         self.attack_start_utime = packets[0].time
         self.attack_end_utime = packets[-1].time

code/Attack/SalityBotnet.py

@@ -7,7 +7,7 @@ from scapy.layers.inet import Ether
 from Attack import BaseAttack
 from Attack.AttackParameters import Parameter as Param
 from Attack.AttackParameters import ParameterTypes
-from ID2TLib.Utility import update_timestamp, get_interval_pps
+from ID2TLib.Utility import update_timestamp, get_interval_pps, handle_most_used_outputs
 
 logging.getLogger("scapy.runtime").setLevel(logging.ERROR)
 # noinspection PyPep8
@@ -26,13 +26,13 @@ class SalityBotnet(BaseAttack.BaseAttack):
                                         "Botnet")
 
         # Define allowed parameters and their type
-        self.supported_params = {
+        self.supported_params.update({
             Param.MAC_SOURCE: ParameterTypes.TYPE_MAC_ADDRESS,
             Param.IP_SOURCE: ParameterTypes.TYPE_IP_ADDRESS,
             Param.INJECT_AT_TIMESTAMP: ParameterTypes.TYPE_FLOAT,
             Param.INJECT_AFTER_PACKET: ParameterTypes.TYPE_PACKET_POSITION,
             Param.PACKETS_PER_SECOND: ParameterTypes.TYPE_FLOAT
-        }
+        })
 
     def init_params(self):
         """
@@ -45,8 +45,7 @@ class SalityBotnet(BaseAttack.BaseAttack):
         # PARAMETERS: initialize with default utilsvalues
         # (values are overwritten if user specifies them)
         most_used_ip_address = self.statistics.get_most_used_ip_address()
-        if isinstance(most_used_ip_address, list):
-            most_used_ip_address = most_used_ip_address[0]
+
         self.add_param_value(Param.IP_SOURCE, most_used_ip_address)
         self.add_param_value(Param.MAC_SOURCE, self.statistics.get_mac_address(most_used_ip_address))
 
@@ -72,7 +71,8 @@ class SalityBotnet(BaseAttack.BaseAttack):
         ip_source = self.get_param_value(Param.IP_SOURCE)
 
         # Pick a DNS server from the background traffic
-        ip_dns_server = self.statistics.process_db_query("SELECT ipAddress FROM ip_protocols WHERE protocolName='DNS' ORDER BY protocolCount DESC LIMIT 1;")
+        ip_dns_server = self.statistics.process_db_query("SELECT ipAddress FROM ip_protocols WHERE protocolName='DNS' AND protocolCount=(SELECT MAX(protocolCount) FROM ip_protocols WHERE protocolName='DNS');")
+        ip_dns_server = handle_most_used_outputs(ip_dns_server)
         if not ip_dns_server or ip_source == ip_dns_server:
             ip_dns_server = self.statistics.get_random_ip_address()
         mac_dns_server = self.statistics.get_mac_address(ip_dns_server)
@@ -127,6 +127,7 @@ class SalityBotnet(BaseAttack.BaseAttack):
 
             packets.append(new_pkt)
 
+        exploit_raw_packets.close()
         # Store timestamp of first packet (for attack label)
         self.attack_start_utime = packets[0].time
         self.attack_end_utime = packets[-1].time

code/CLI.py

@@ -61,7 +61,8 @@ class CLI(object):
                             help='query the statistics database. If no query is provided, the application enters query mode.')
         parser.add_argument('-t', '--extraTests', help='perform extra tests on the input pcap file, including calculating IP entropy'
                                                        'in interval-wise, TCP checksum, and checking payload availability.', action='store_true')
-
+        parser.add_argument('-S', '--randomSeed', action='append', help='sets random seed for testing or benchmarking',
+                            nargs='+', default=[])
 
         # Attack arguments
         parser.add_argument('-a', '--attack', metavar="ATTACK", action='append',
@@ -141,10 +142,14 @@ class CLI(object):
         if self.args.plot is not None:
             controller.create_statistics_plot(self.args.plot)
 
+        # Check random seed
+        if not isinstance(self.args.randomSeed, list):
+            self.args.randomSeed = [self.args.randomSeed]
+
         # Process attack(s) with given attack params
         if self.args.attack is not None:
             # If attack is present, load attack with params
-            controller.process_attacks(self.args.attack)
+            controller.process_attacks(self.args.attack, self.args.randomSeed)
 
         # Parameter -q without arguments was given -> go into query loop
         if self.args.query == [None]:
@@ -153,6 +158,7 @@ class CLI(object):
         elif self.args.query is not None:
             controller.process_db_queries(self.args.query, True)
 
+
 def main(args):
     """
     Creates a new CLI object and invokes the arguments parsing.
@@ -163,6 +169,7 @@ def main(args):
     # Check arguments
     cli.parse_arguments(args)
 
+
 # Uncomment to enable calling by terminal
 if __name__ == '__main__':
     main(sys.argv[1:])

code/ID2TLib/AttackController.py

@@ -20,11 +20,21 @@ class AttackController:
 
         self.current_attack = None
         self.added_attacks = []
+        self.seed = None
 
-    def create_attack(self, attack_name: str):
+    def set_seed(self, seed: int):
+        """
+        Sets global seed.
+
+        :param seed: random seed
+        """
+        self.seed = seed
+
+    def create_attack(self, attack_name: str, seed=None):
         """
         Creates dynamically a new class instance based on the given attack_name.
         :param attack_name: The name of the attack, must correspond to the attack's class name.
+        :param seed: random seed for param generation
         :return: None
         """
         print("\nCreating attack instance of \033[1m" + attack_name + "\033[0m")
@@ -36,6 +46,8 @@ class AttackController:
         self.current_attack = attack_class()
         # Initialize the parameters of the attack with defaults or user supplied values.
         self.current_attack.set_statistics(self.statistics)
+        if seed is not None:
+            self.current_attack.set_seed(seed=seed)
         self.current_attack.init_params()
         # Record the attack
         self.added_attacks.append(self.current_attack)
@@ -48,10 +60,11 @@ class AttackController:
         :param params: The parameters for attack customization, see attack class for supported params.
         :return: The file path to the created pcap file.
         """
-        self.create_attack(attack)
+        self.create_attack(attack, self.seed)
 
-        # Add attack parameters if provided
         print("Validating and adding attack parameters.")
+
+        # Add attack parameters if provided
         params_dict = []
         if isinstance(params, list) and params:
             # Convert attack param list into dictionary

code/ID2TLib/Controller.py

@@ -19,6 +19,7 @@ class Controller:
         self.pcap_dest_path = ''
         self.written_pcaps = []
         self.do_extra_tests = do_extra_tests
+        self.seed = None
 
         # Initialize class instances
         print("Input file: %s" % self.pcap_src_path)
@@ -40,7 +41,7 @@ class Controller:
         """
         self.statistics.load_pcap_statistics(flag_write_file, flag_recalculate_stats, flag_print_statistics)
 
-    def process_attacks(self, attacks_config: list):
+    def process_attacks(self, attacks_config: list, seeds=[]):
         """
         Creates the attack based on the attack name and the attack parameters given in the attacks_config. The
         attacks_config is a list of attacks, e.g.
@@ -50,9 +51,13 @@ class Controller:
         :param attacks_config: A list of attacks with their attack parameters.
         """
         # load attacks sequentially
+        i = 0
         for attack in attacks_config:
+            if len(seeds) > i:
+                self.attack_controller.set_seed(seed=seeds[i][0])
             temp_attack_pcap = self.attack_controller.process_attack(attack[0], attack[1:])
             self.written_pcaps.append(temp_attack_pcap)
+            i += 1
 
         # merge attack pcaps to get single attack pcap
         if len(self.written_pcaps) > 1:
@@ -71,6 +76,15 @@ class Controller:
         print("Merging base pcap with single attack pcap...", end=" ")
         sys.stdout.flush()  # force python to print text immediately
         self.pcap_dest_path = self.pcap_file.merge_attack(attacks_pcap_path)
+
+        tmp_path_tuple = self.pcap_dest_path.rpartition("/")
+        result_dir = tmp_path_tuple[0] + tmp_path_tuple[1] + "ID2T_results/"
+        result_path = result_dir + tmp_path_tuple[2]
+
+        os.makedirs(result_dir, exist_ok=True)
+        os.rename(self.pcap_dest_path, result_path)
+        self.pcap_dest_path = result_path
+
         print("done.")
 
         # delete intermediate PCAP files

code/ID2TLib/LabelManager.py

@@ -28,7 +28,7 @@ class LabelManager:
         self.labels = list()
 
         if filepath_pcap is not None:
-            self.label_file_path = filepath_pcap.strip('.pcap') + '_labels.xml'
+            self.label_file_path = os.path.splitext(filepath_pcap)[0] + '_labels.xml'
             # only load labels if label file is existing
             if os.path.exists(self.label_file_path):
                 self.load_labels()
@@ -83,7 +83,7 @@ class LabelManager:
             return timestamp_root
 
         if filepath is not None:
-            self.label_file_path = filepath.strip('.pcap') + '_labels.xml'
+            self.label_file_path = os.path.splitext(filepath)[0] + '_labels.xml'
 
         # Generate XML
         doc = Document()

code/ID2TLib/SMBLib.py

@@ -1,6 +1,6 @@
 from os import urandom
 from binascii import b2a_hex
-from random import random
+from random import choice
 
 from ID2TLib.Utility import check_platform, get_filetime_format, get_rnd_boot_time
 
@@ -60,7 +60,7 @@ def get_smb_version(platform: str):
     """
     check_platform(platform)
     if platform is "linux":
-        return random.choice(list(smb_versions_per_samba.values()))
+        return choice(list(smb_versions_per_samba.values()))
     elif platform is "macos":
         return "2.1"
     else:

code/ID2TLib/Statistics.py

@@ -3,13 +3,15 @@ from math import sqrt, ceil, log
 
 import os
 import time
+import random
 import ID2TLib.libpcapreader as pr
 import matplotlib
 
-matplotlib.use('Agg')
+matplotlib.use('Agg', force=True)
 import matplotlib.pyplot as plt
 from ID2TLib.PcapFile import PcapFile
 from ID2TLib.StatsDatabase import StatsDatabase
+from ID2TLib.Utility import handle_most_used_outputs
 
 
 class Statistics:
@@ -477,7 +479,7 @@ class Statistics:
         """
         :return: The IP address/addresses with the highest sum of packets sent and received
         """
-        return self.process_db_query("most_used(ipAddress)")
+        return handle_most_used_outputs(self.process_db_query("most_used(ipAddress)"))
 
     def get_ttl_distribution(self, ipAddress: str):
         result = self.process_db_query('SELECT ttlValue, ttlCount from ip_ttl WHERE ipAddress="' + ipAddress + '"')
@@ -511,13 +513,16 @@ class Statistics:
         :return: A randomly chosen IP address from the dataset or iff param count is greater than one, a list of randomly
          chosen IP addresses
         """
+        ip_address_list = self.process_db_query("all(ipAddress)")
         if count == 1:
-            return self.process_db_query("random(all(ipAddress))")
+            return random.choice(ip_address_list)
         else:
-            ip_address_list = []
+            result_list = []
             for i in range(0, count):
-                ip_address_list.append(self.process_db_query("random(all(ipAddress))"))
-            return ip_address_list
+                random_ip = random.choice(ip_address_list)
+                result_list.append(random_ip)
+                ip_address_list.remove(random_ip)
+            return result_list
 
     def get_ip_address_from_mac(self, macAddress: str):
         """
@@ -538,9 +543,15 @@ class Statistics:
         :return: The TCP MSS value used by the IP address, or if the IP addresses never specified a MSS,
         then None is returned
         """
-        mss_value = self.process_db_query('SELECT mssValue from tcp_mss WHERE ipAddress="' + ipAddress + '" ORDER BY mssCount DESC LIMIT 1')
+        mss_value = self.process_db_query('SELECT mssValue from tcp_mss WHERE ipAddress="' + ipAddress + '" AND mssCount == (SELECT MAX(mssCount) from tcp_mss WHERE ipAddress="' + ipAddress + '")')
         if isinstance(mss_value, int):
             return mss_value
+        elif isinstance(mss_value, list):
+            if len(mss_value) == 0:
+                return None
+            else:
+                mss_value.sort()
+                return mss_value[0]
         else:
             return None
 
@@ -551,12 +562,31 @@ class Statistics:
         then None is returned
         """
         ttl_value = self.process_db_query(
-            'SELECT ttlValue from ip_ttl WHERE ipAddress="' + ipAddress + '" ORDER BY ttlCount DESC LIMIT 1')
+            'SELECT ttlValue from ip_ttl WHERE ipAddress="' + ipAddress + '" AND ttlCount == (SELECT MAX(ttlCount) from ip_ttl WHERE ipAddress="' + ipAddress + '")')
         if isinstance(ttl_value, int):
             return ttl_value
+        elif isinstance(ttl_value, list):
+            if len(ttl_value) == 0:
+                return None
+            else:
+                ttl_value.sort()
+                return ttl_value[0]
         else:
             return None
 
+    def get_rnd_win_size(self, pkts_num):
+        """
+        :param pkts_num: maximum number of window sizes, that should be returned
+        :return: A list of randomly chosen window sizes with given length.
+        """
+        sql_return = self.process_db_query("SELECT DISTINCT winSize FROM tcp_win ORDER BY winsize ASC;")
+        if not isinstance(sql_return, list):
+            return [sql_return]
+        result = []
+        for i in range(0, min(pkts_num, len(sql_return))):
+            result.append(random.choice(sql_return))
+            sql_return.remove(result[i])
+        return result
 
     def get_statistics_database(self):
         """

code/ID2TLib/StatsDatabase.py

@@ -169,31 +169,34 @@ class StatsDatabase:
         """
         # Definition of SQL queries associated to named queries
         named_queries = {
-            "most_used.ipaddress": "SELECT ipAddress FROM ip_statistics WHERE (pktsSent+pktsReceived) == (SELECT MAX(pktsSent+pktsReceived) from ip_statistics) LIMIT 1",
-            "most_used.macaddress": "SELECT * FROM (SELECT macAddress, COUNT(*) as occ from ip_mac GROUP BY macAddress ORDER BY occ DESC) WHERE occ=(SELECT COUNT(*) as occ from ip_mac GROUP BY macAddress ORDER BY occ DESC LIMIT 1)",
-            "most_used.portnumber": "SELECT portNumber, COUNT(portNumber) as cntPort FROM ip_ports GROUP BY portNumber HAVING cntPort=(SELECT MAX(cntPort) from (SELECT portNumber, COUNT(portNumber) as cntPort FROM ip_ports GROUP BY portNumber))",
-            "most_used.protocolname": "SELECT protocolName, COUNT(protocolCount) as countProtocols FROM ip_protocols GROUP BY protocolName HAVING countProtocols=(SELECT COUNT(protocolCount) as cnt FROM ip_protocols GROUP BY protocolName ORDER BY cnt DESC LIMIT 1)",
-            "most_used.ttlvalue": "SELECT ttlValue FROM ip_ttl GROUP BY ttlValue ORDER BY SUM(ttlCount) DESC LIMIT 1",
-            "most_used.mssvalue": "SELECT mssValue FROM tcp_mss GROUP BY mssValue ORDER BY SUM(mssCount) DESC LIMIT 1",
-            "most_used.winsize": "SELECT winSize FROM tcp_win GROUP BY winSize ORDER BY SUM(winCount) DESC LIMIT 1",
-            "most_used.ipclass": "SELECT ipClass FROM ip_statistics GROUP BY ipClass ORDER BY COUNT(*) DESC LIMIT 1",
-            "least_used.ipaddress": "SELECT ipAddress FROM ip_statistics WHERE (pktsSent+pktsReceived) == (SELECT MIN(pktsSent+pktsReceived) from ip_statistics)",
-            "least_used.macaddress": "SELECT * FROM (SELECT macAddress, COUNT(*) as occ from ip_mac GROUP BY macAddress ORDER BY occ ASC) WHERE occ=(SELECT COUNT(*) as occ from ip_mac GROUP BY macAddress ORDER BY occ ASC LIMIT 1)",
-            "least_used.portnumber": "SELECT portNumber, COUNT(portNumber) as cntPort FROM ip_ports GROUP BY portNumber HAVING cntPort=(SELECT MIN(cntPort) from (SELECT portNumber, COUNT(portNumber) as cntPort FROM ip_ports GROUP BY portNumber))",
-            "least_used.protocolname": "SELECT protocolName, COUNT(protocolCount) as countProtocols FROM ip_protocols GROUP BY protocolName HAVING countProtocols=(SELECT COUNT(protocolCount) as cnt FROM ip_protocols GROUP BY protocolName ORDER BY cnt ASC LIMIT 1)",
-            "least_used.ttlvalue": "SELECT ttlValue FROM ip_ttl WHERE ttlCount == (SELECT MIN(ttlCount) FROM ip_ttl)",
+            "most_used.ipaddress": "SELECT ipAddress FROM ip_statistics WHERE (pktsSent+pktsReceived) == (SELECT MAX(pktsSent+pktsReceived) from ip_statistics) ORDER BY ipAddress ASC",
+            "most_used.macaddress": "SELECT macAddress FROM (SELECT macAddress, COUNT(*) as occ from ip_mac GROUP BY macAddress) WHERE occ=(SELECT COUNT(*) as occ from ip_mac GROUP BY macAddress ORDER BY occ DESC LIMIT 1) ORDER BY macAddress ASC",
+            "most_used.portnumber": "SELECT portNumber FROM ip_ports GROUP BY portNumber HAVING COUNT(portNumber)=(SELECT MAX(cntPort) from (SELECT portNumber, COUNT(portNumber) as cntPort FROM ip_ports GROUP BY portNumber)) ORDER BY portNumber ASC",
+            "most_used.protocolname": "SELECT protocolName FROM ip_protocols GROUP BY protocolName HAVING COUNT(protocolCount)=(SELECT COUNT(protocolCount) as cnt FROM ip_protocols GROUP BY protocolName ORDER BY cnt DESC LIMIT 1) ORDER BY protocolName ASC",
+            "most_used.ttlvalue": "SELECT ttlValue FROM (SELECT ttlValue, SUM(ttlCount) as occ FROM ip_ttl GROUP BY ttlValue) WHERE occ=(SELECT SUM(ttlCount) as occ FROM ip_ttl GROUP BY ttlValue ORDER BY occ DESC LIMIT 1) ORDER BY ttlValue ASC",
+            "most_used.mssvalue": "SELECT mssValue FROM (SELECT mssValue, SUM(mssCount) as occ FROM tcp_mss GROUP BY mssValue) WHERE occ=(SELECT SUM(mssCount) as occ FROM tcp_mss GROUP BY mssValue ORDER BY occ DESC LIMIT 1) ORDER BY mssValue ASC",
+            "most_used.winsize": "SELECT winSize FROM (SELECT winSize, SUM(winCount) as occ FROM tcp_win GROUP BY winSize) WHERE occ=(SELECT SUM(winCount) as occ FROM tcp_win GROUP BY winSize ORDER BY occ DESC LIMIT 1) ORDER BY winSize ASC",
+            "most_used.ipclass": "SELECT ipClass FROM (SELECT ipClass, COUNT(*) as occ from ip_statistics GROUP BY ipClass ORDER BY occ DESC) WHERE occ=(SELECT COUNT(*) as occ from ip_statistics GROUP BY ipClass ORDER BY occ DESC LIMIT 1) ORDER BY ipClass ASC",
+            #FIXME ORDER BY ASC ? check queries for os dependency!!
+            "least_used.ipaddress": "SELECT ipAddress FROM ip_statistics WHERE (pktsSent+pktsReceived) == (SELECT MIN(pktsSent+pktsReceived) from ip_statistics) ORDER BY ipAddress ASC",
+            "least_used.macaddress": "SELECT macAddress FROM (SELECT macAddress, COUNT(*) as occ from ip_mac GROUP BY macAddress) WHERE occ=(SELECT COUNT(*) as occ from ip_mac GROUP BY macAddress ORDER BY occ ASC LIMIT 1) ORDER BY macAddress ASC",
+            "least_used.portnumber": "SELECT portNumber FROM ip_ports GROUP BY portNumber HAVING COUNT(portNumber)=(SELECT MIN(cntPort) from (SELECT portNumber, COUNT(portNumber) as cntPort FROM ip_ports GROUP BY portNumber)) ORDER BY portNumber ASC",
+            "least_used.protocolname": "SELECT protocolName FROM ip_protocols GROUP BY protocolName HAVING COUNT(protocolCount)=(SELECT COUNT(protocolCount) as cnt FROM ip_protocols GROUP BY protocolName ORDER BY cnt ASC LIMIT 1) ORDER BY protocolName ASC",
+            "least_used.ttlvalue": "SELECT ttlValue FROM (SELECT ttlValue, SUM(ttlCount) as occ FROM ip_ttl GROUP BY ttlValue) WHERE occ=(SELECT SUM(ttlCount) as occ FROM ip_ttl GROUP BY ttlValue ORDER BY occ ASC LIMIT 1) ORDER BY ttlValue ASC",
+            "least_used.mssvalue": "SELECT mssValue FROM (SELECT mssValue, SUM(mssCount) as occ FROM tcp_mss GROUP BY mssValue) WHERE occ=(SELECT SUM(mssCount) as occ FROM tcp_mss GROUP BY mssValue ORDER BY occ ASC LIMIT 1) ORDER BY mssValue ASC",
+            "least_used.winsize": "SELECT winSize FROM (SELECT winSize, SUM(winCount) as occ FROM tcp_win GROUP BY winSize) WHERE occ=(SELECT SUM(winCount) as occ FROM tcp_win GROUP BY winSize ORDER BY occ ASC LIMIT 1) ORDER BY winSize ASC",
             "avg.pktsreceived": "SELECT avg(pktsReceived) from ip_statistics",
             "avg.pktssent": "SELECT avg(pktsSent) from ip_statistics",
             "avg.kbytesreceived": "SELECT avg(kbytesReceived) from ip_statistics",
             "avg.kbytessent": "SELECT avg(kbytesSent) from ip_statistics",
             "avg.ttlvalue": "SELECT avg(ttlValue) from ip_ttl",
             "avg.mss": "SELECT avg(mssValue) from tcp_mss",
-            "all.ipaddress": "SELECT ipAddress from ip_statistics",
-            "all.ttlvalue": "SELECT DISTINCT ttlValue from ip_ttl",
-            "all.mss": "SELECT DISTINCT mssValue from tcp_mss",
-            "all.macaddress": "SELECT DISTINCT macAddress from ip_mac",
-            "all.portnumber": "SELECT DISTINCT portNumber from ip_ports",
-            "all.protocolname": "SELECT DISTINCT protocolName from ip_protocols"}
+            "all.ipaddress": "SELECT ipAddress from ip_statistics ORDER BY ipAddress ASC",
+            "all.ttlvalue": "SELECT DISTINCT ttlValue from ip_ttl ORDER BY ttlValue ASC",
+            "all.mss": "SELECT DISTINCT mssValue from tcp_mss ORDER BY mssValue ASC",
+            "all.macaddress": "SELECT DISTINCT macAddress from ip_mac ORDER BY macAddress ASC",
+            "all.portnumber": "SELECT DISTINCT portNumber from ip_ports ORDER BY portNumber ASC",
+            "all.protocolname": "SELECT DISTINCT protocolName from ip_protocols ORDER BY protocolName ASC"}
 
         # Retrieve values by selectors, if given, reduce results by extractor
         last_result = 0

code/ID2TLib/TestLibrary.py

@@ -0,0 +1,126 @@
+import hashlib
+import os
+import random
+
+import ID2TLib.Utility as Util
+
+test_resource_dir = Util.TEST_DIR
+test_pcap = Util.TEST_DIR + "reference_1998.pcap"
+test_pcap_ips = ["10.0.2.15", "52.85.173.182"]
+test_pcap_empty = []
+
+"""
+helper functions for ID2TAttackTest
+"""
+
+
+def get_sha256(file):
+    """
+    Generates a sha256 checksum from file
+
+    :param file: absolute path to file
+    :return: sha256 checksum
+    """
+    sha = hashlib.sha256()
+    with open(file, 'rb') as f:
+        while True:
+            data = f.read(0x100000)
+            if not data:
+                break
+            sha.update(data)
+    f.close()
+    return sha.hexdigest()
+
+
+def clean_up(controller):
+    """
+    Removes the output files from a given controller
+
+    :param controller: controller which created output files
+    """
+    os.remove(controller.pcap_dest_path)
+    os.remove(controller.label_manager.label_file_path)
+
+
+def rename_test_result_files(controller, caller_function: str, attack_sub_dir=False, test_sub_dir=False):
+    """
+    :param controller: controller which created output files
+    :param caller_function: the name of the function which called the generic test
+    :param attack_sub_dir: create sub-directory for each attack-class if True
+    :param test_sub_dir: create sub-directory for each test-function/case if True
+    """
+    tmp_path_tuple = controller.pcap_dest_path.rpartition("_")
+    result_pcap_path = tmp_path_tuple[0] + tmp_path_tuple[1] + caller_function + "_" + tmp_path_tuple[2]
+
+    tmp_label_path_tuple = controller.label_manager.label_file_path.rpartition("_")
+    tmp_path_tuple = tmp_label_path_tuple[0].rpartition("_")
+    result_labels_path = tmp_path_tuple[0] + tmp_path_tuple[1] + caller_function + "_" + tmp_path_tuple[2]
+    result_labels_path = result_labels_path + tmp_label_path_tuple[1] + tmp_label_path_tuple[2]
+
+    if attack_sub_dir:
+        caller_attack = caller_function.replace("test_", "").partition("_")[0]
+        tmp_dir_tuple = result_pcap_path.rpartition("/")
+        result_dir = tmp_dir_tuple[0] + tmp_dir_tuple[1] + caller_attack + "/"
+        result_pcap_path = result_dir + tmp_dir_tuple[2]
+        os.makedirs(result_dir, exist_ok=True)
+
+        tmp_dir_tuple = result_labels_path.rpartition("/")
+        result_labels_path = result_dir + tmp_dir_tuple[2]
+
+    if test_sub_dir:
+        tmp_dir_tuple = result_pcap_path.rpartition("/")
+        result_dir = tmp_dir_tuple[0] + tmp_dir_tuple[1] + (caller_function.replace("test_", "")) + "/"
+        result_pcap_path = result_dir + tmp_dir_tuple[2]
+        os.makedirs(result_dir, exist_ok=True)
+
+        tmp_dir_tuple = result_labels_path.rpartition("/")
+        result_labels_path = result_dir + tmp_dir_tuple[2]
+
+    os.rename(controller.pcap_dest_path, result_pcap_path)
+    controller.pcap_dest_path = result_pcap_path
+
+    os.rename(controller.label_manager.label_file_path, result_labels_path)
+    controller.label_manager.label_file_path = result_labels_path
+
+
+"""
+function patches for unittests
+"""
+
+
+def get_bytes(count, ignore):
+    """
+    unittest patch for get_rnd_bytes (ID2TLib.Utility.py)
+
+    :param count: count of requested bytes
+    :param ignore: <not used>
+    :return: a count of As
+    """
+    return b'A' * count
+
+
+def get_x86_nop(count, side_effect_free, char_filter):
+    """
+    unittest patch for get_rnd_x86_nop (ID2TLib.Utility.py)
+
+    :param count: count of requested nops
+    :param side_effect_free: <not used>
+    :param char_filter: <not used>
+    :return: a count of \x90
+    """
+    return b'\x90' * count
+
+
+def get_attacker_config(ip_source_list, ipAddress: str):
+    """
+    unittest patch for get_attacker_config (ID2TLib.Utility.py)
+
+    :param ip_source_list: List of source IPs
+    :param ipAddress: The IP address of the attacker
+    :return: A tuple consisting of (port, ttlValue)
+    """
+    next_port = random.randint(0, 2 ** 16 - 1)
+    ttl = random.randint(1, 255)
+
+    return next_port, ttl
+

code/ID2TLib/Utility.py

@@ -1,10 +1,19 @@
 import ipaddress
+import os
 
 from random import randint, uniform
 from os import urandom
 from datetime import datetime
 from calendar import timegm
 from lea import Lea
+from scipy.stats import gamma
+from scapy.layers.inet import RandShort
+
+CODE_DIR = os.path.dirname(os.path.abspath(__file__)) + "/../"
+ROOT_DIR = CODE_DIR + "../"
+RESOURCE_DIR = ROOT_DIR + "resources/"
+TEST_DIR = RESOURCE_DIR + "test/"
+
 
 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,
@@ -18,6 +27,9 @@ x86_pseudo_nops = {b'\x97', b'\x96', b'\x95', b'\x93', b'\x92', b'\x91', b'\x99'
                    b'\x5b', b'\x59', b'\x5f', b'\x5a', b'\x5e', b'\xd6'}
 forbidden_chars = [b'\x00', b'\x0a', b'\x0d']
 
+attacker_port_mapping = {}
+attacker_ttl_mapping = {}
+
 
 def update_timestamp(timestamp, pps, delay=0):
     """
@@ -188,9 +200,9 @@ def get_rnd_x86_nop(count=1, side_effect_free=False, char_filter=set()):
     :return: Random x86 NOP bytestring
     """
     result = b''
-    nops = x86_nops
+    nops = x86_nops.copy()
     if not side_effect_free:
-        nops |= x86_pseudo_nops
+        nops |= x86_pseudo_nops.copy()
 
     if not isinstance(char_filter, set):
         char_filter = set(char_filter)
@@ -220,6 +232,18 @@ def get_rnd_bytes(count=1, ignore=None):
     return result
 
 
+def check_payload_len(payload_len: int, limit: int):
+    """
+    Checks if the len of the payload exceeds a given limit
+    :param payload_len: The length of the payload
+    :param limit: The limit of the length of the payload which is allowed
+    """
+
+    if payload_len > limit:
+        print("\nCustom payload too long: ", payload_len, " bytes. Should be a maximum of ", limit, " bytes.")
+        exit(1)
+
+
 def get_bytes_from_file(filepath):
     """
     Converts the content of a file into its byte representation
@@ -249,16 +273,19 @@ def get_bytes_from_file(filepath):
                 result_bytes = bytes.fromhex(content)
             except ValueError:
                 print("\nERROR: Content of file is not all hexadecimal.")
+                file.close()
                 exit(1)
         elif header == "str":
-            result_bytes = content.encode()
+            result_bytes = content.strip().encode()
         else:
             print("\nERROR: Invalid header found: " + header + ". Try 'hex' or 'str' followed by endline instead.")
+            file.close()
             exit(1)
 
         for forbidden_char in forbidden_chars:
             if forbidden_char in result_bytes:
                 print("\nERROR: Forbidden character found in payload: ", forbidden_char)
+                file.close()
                 exit(1)
 
         file.close()
@@ -267,3 +294,56 @@ def get_bytes_from_file(filepath):
     except FileNotFoundError:
         print("\nERROR: File not found: ", filepath)
         exit(1)
+
+
+def handle_most_used_outputs(most_used_x):
+    """
+    :param most_used_x: Element or list (e.g. from SQL-query output) which should only be one element
+    :return: most_used_x if it's not a list. The first element of most_used_x after being sorted if it's a list.
+    None if that list is empty.
+    """
+    if isinstance(most_used_x, list):
+        if len(most_used_x) == 0:
+            return None
+        most_used_x.sort()
+        return most_used_x[0]
+    else:
+        return most_used_x
+
+
+def get_attacker_config(ip_source_list, ipAddress: str):
+    """
+    Returns the attacker configuration depending on the IP address, this includes the port for the next
+    attacking packet and the previously used (fixed) TTL value.
+    :param ip_source_list: List of source IPs
+    :param ipAddress: The IP address of the attacker
+    :return: A tuple consisting of (port, ttlValue)
+    """
+    # Gamma distribution parameters derived from MAWI 13.8G dataset
+    alpha, loc, beta = (2.3261710235, -0.188306914406, 44.4853123884)
+    gd = gamma.rvs(alpha, loc=loc, scale=beta, size=len(ip_source_list))
+
+    # Determine port
+    port = attacker_port_mapping.get(ipAddress)
+    if port is not None:  # use next port
+        next_port = attacker_port_mapping.get(ipAddress) + 1
+        if next_port > (2 ** 16 - 1):
+            next_port = 1
+    else:  # generate starting port
+        next_port = RandShort()
+    attacker_port_mapping[ipAddress] = next_port
+    # Determine TTL value
+    ttl = attacker_ttl_mapping.get(ipAddress)
+    if ttl is None:  # determine TTL value
+        is_invalid = True
+        pos = ip_source_list.index(ipAddress)
+        pos_max = len(gd)
+        while is_invalid:
+            ttl = int(round(gd[pos]))
+            if 0 < ttl < 256:  # validity check
+                is_invalid = False
+            else:
+                pos = index_increment(pos, pos_max)
+        attacker_ttl_mapping[ipAddress] = ttl
+    # return port and TTL
+    return next_port, ttl

code/Test/ID2TAttackTest.py

@@ -0,0 +1,46 @@
+import unittest
+import inspect
+
+import ID2TLib.Controller as Ctrl
+import ID2TLib.TestLibrary as Lib
+
+
+class ID2TAttackTest(unittest.TestCase):
+    """
+    Generic Test Class for ID2T attacks based on unittest.TestCase.
+    """
+
+    def checksum_test(self, attack_args, sha256_checksum, seed=5, cleanup=True, pcap=Lib.test_pcap,
+                      flag_write_file=False, flag_recalculate_stats=False, flag_print_statistics=False,
+                      attack_sub_dir=True, test_sub_dir=True):
+        """
+        Runs the attack against a given sha256 checksum.
+
+        :param attack_args: A list of attacks with their attack parameters (as defined in Controller.process_attacks).
+        :param sha256_checksum: The checksum to verify the result pcap.
+        :param seed: A random seed to keep random values static (care for count and order of random generation).
+        :param cleanup: Clean up attack output after testing.
+        :param pcap: The input pcap for the attack.
+        :param flag_write_file: Writes the statistics to a file.
+        :param flag_recalculate_stats: Forces the recalculation of statistics.
+        :param flag_print_statistics: Prints the statistics on the terminal.
+        :param attack_sub_dir: create sub-directory for each attack-class if True
+        :param test_sub_dir: create sub-directory for each test-function/case if True
+        """
+
+        controller = Ctrl.Controller(pcap_file_path=pcap, do_extra_tests=False)
+        controller.load_pcap_statistics(flag_write_file, flag_recalculate_stats, flag_print_statistics)
+        controller.process_attacks(attack_args, [[seed]])
+
+        caller_function = inspect.stack()[1].function
+
+        try:
+            self.assertEqual(sha256_checksum, Lib.get_sha256(controller.pcap_dest_path))
+        except self.failureException:
+            Lib.rename_test_result_files(controller, caller_function, attack_sub_dir, test_sub_dir)
+            raise
+
+        if cleanup:
+            Lib.clean_up(controller)
+        else:
+            Lib.rename_test_result_files(controller, caller_function, attack_sub_dir, test_sub_dir)

code/Test/test_BaseAttack.py

@@ -0,0 +1,182 @@
+import unittest
+import Attack.BaseAttack as BA
+
+# TODO: improve coverage
+
+
+class TestBaseAttack(unittest.TestCase):
+
+    def test_is_mac_address_valid(self):
+        self.assertTrue(BA.BaseAttack._is_mac_address("00:80:41:ae:fd:7e"))
+
+    def test_is_mac_address_invalid(self):
+        self.assertFalse(BA.BaseAttack._is_mac_address("00:80:41:aec:fd:7e"))
+
+    def test_is_mac_address_empty(self):
+        self.assertFalse(BA.BaseAttack._is_mac_address(""))
+
+    def test_is_mac_address_minus_valid(self):
+        self.assertTrue(BA.BaseAttack._is_mac_address("00-80-41-ae-fd-7e"))
+
+    def test_is_mac_address_minus_invalid(self):
+        self.assertFalse(BA.BaseAttack._is_mac_address("00-80-41-aec-fd-7e"))
+
+    def test_is_mac_address_list_valid(self):
+        self.assertTrue(BA.BaseAttack._is_mac_address(["00:80:41:ae:fd:7e", "00-80-41-ae-fd-7e"]))
+
+    def test_is_mac_address_list_invalid(self):
+        self.assertFalse(BA.BaseAttack._is_mac_address(["00:80:41:aec:fd:7e", "00-80-41-aec-fd-7e"]))
+
+    def test_is_ip_address_empty(self):
+        self.assertFalse(BA.BaseAttack._is_ip_address("")[0])
+
+    def test_is_ip_address_v4_valid(self):
+        self.assertTrue(BA.BaseAttack._is_ip_address("192.168.178.1")[0])
+
+    def test_is_ip_address_v4_invalid(self):
+        self.assertFalse(BA.BaseAttack._is_ip_address("192.1689.178.1")[0])
+
+    def test_is_ip_address_v6_valid(self):
+        self.assertTrue(BA.BaseAttack._is_ip_address("2001:0db8:85a3:08d3:1319:8a2e:0370:7344")[0])
+
+    def test_is_ip_address_v6_invalid(self):
+        self.assertFalse(BA.BaseAttack._is_ip_address("2001:0db8:85a3:08d3X:1319:8a2e:0370:7344")[0])
+
+    def test_is_ip_address_v6_shortened_valid(self):
+        self.assertTrue(BA.BaseAttack._is_ip_address("2001:0db8:85a3:08d3:1319::0370:7344")[0])
+
+    def test_is_ip_address_v6_shortened_invalid(self):
+        self.assertFalse(BA.BaseAttack._is_ip_address("2001::85a3:08d3X::8a2e:0370:7344")[0])
+
+    def test_is_ip_address_list_valid(self):
+        self.assertTrue(BA.BaseAttack._is_ip_address(["192.168.178.1", "192.168.178.10"])[0])
+
+    def test_is_ip_address_list_invalid(self):
+        self.assertFalse(BA.BaseAttack._is_ip_address(["192.1689.178.1", "192.168.178.10"])[0])
+
+    def test_is_ip_address_comma_list_valid(self):
+        self.assertTrue(BA.BaseAttack._is_ip_address("192.168.178.1,192.168.178.10")[0])
+
+    def test_is_ip_address_comma_list_invalid(self):
+        self.assertFalse(BA.BaseAttack._is_ip_address("192.168.178.1,192.1689.178.10")[0])
+
+    def test_is_port_none(self):
+        self.assertFalse(BA.BaseAttack._is_port(None))
+
+    def test_is_port_empty(self):
+        self.assertFalse(BA.BaseAttack._is_port(""))
+
+    def test_is_port_empty_list(self):
+        self.assertFalse(BA.BaseAttack._is_port([]))
+
+    def test_is_port_valid(self):
+        self.assertTrue(BA.BaseAttack._is_port(5000))
+
+    def test_is_port_invalid(self):
+        self.assertFalse(BA.BaseAttack._is_port(70000))
+
+    def test_is_port_string_valid(self):
+        self.assertTrue(BA.BaseAttack._is_port("5000"))
+
+    def test_is_port_string_invalid(self):
+        self.assertFalse(BA.BaseAttack._is_port("70000"))
+
+    def test_is_port_string_comma_valid(self):
+        self.assertTrue(BA.BaseAttack._is_port("5000, 4000, 3000"))
+
+    def test_is_port_string_comma_ivalid(self):
+        self.assertFalse(BA.BaseAttack._is_port("5000, 70000, 3000"))
+
+    def test_is_port_valid_list(self):
+        self.assertTrue(BA.BaseAttack._is_port([5000, 4000, 3000]))
+
+    def test_is_port_invalid_list(self):
+        self.assertFalse(BA.BaseAttack._is_port([5000, 70000, 0]))
+
+    def test_is_port_valid_string_list(self):
+        self.assertTrue(BA.BaseAttack._is_port(["5000", "4000", "3000"]))
+
+    def test_is_port_invalid_string_list(self):
+        self.assertFalse(BA.BaseAttack._is_port(["5000", "70000", "0"]))
+
+    def test_is_port_range_valid(self):
+        self.assertTrue(BA.BaseAttack._is_port("3000-5000"))
+
+    def test_is_port_range_invalid(self):
+        self.assertFalse(BA.BaseAttack._is_port("0-70000"))
+
+    def test_is_port_range_dots_valid(self):
+        self.assertTrue(BA.BaseAttack._is_port("3000...5000"))
+
+    def test_is_port_range_dots_invalid(self):
+        self.assertFalse(BA.BaseAttack._is_port("0...70000"))
+
+    def test_is_port_range_list_valid(self):
+        self.assertTrue(BA.BaseAttack._is_port(["3000-5000", "6000-7000"]))
+
+    def test_is_port_range_list_invalid(self):
+        self.assertFalse(BA.BaseAttack._is_port(["0-70000", "6000-7000"]))
+
+    def test_is_timestamp_valid(self):
+        self.assertTrue(BA.BaseAttack._is_timestamp("2018-01-25 23:54:00"))
+
+    def test_is_timestamp_invalid(self):
+        self.assertFalse(BA.BaseAttack._is_timestamp("20-0100-125 23c:54x:00a"))
+
+    def test_is_boolean_invalid(self):
+        self.assertFalse(BA.BaseAttack._is_boolean("42")[0])
+
+    def test_is_boolean_valid(self):
+        self.assertTrue(BA.BaseAttack._is_boolean(True))
+        self.assertTrue(BA.BaseAttack._is_boolean(False))
+
+    def test_is_boolean_valid_strings(self):
+        for value in {"y", "yes", "t", "true", "on", "1", "n", "no", "f", "false", "off", "0"}:
+            with self.subTest(value=value):
+                self.assertTrue(BA.BaseAttack._is_boolean(value))
+
+    def test_is_float_valid(self):
+        self.assertTrue(BA.BaseAttack._is_float(50.67)[0])
+
+    def test_is_float_invalid(self):
+        self.assertFalse(BA.BaseAttack._is_float("invalid")[0])
+
+    def test_is_domain_valid(self):
+        self.assertTrue(BA.BaseAttack._is_domain("foo://example.com:8042/over/there?name=ferret"))
+
+    def test_is_domain_invalid(self):
+        self.assertFalse(BA.BaseAttack._is_domain("this is not a valid domain, I guess, maybe, let's find out."))
+
+    def test_is_valid_ipaddress_valid(self):
+        self.assertTrue(BA.BaseAttack.is_valid_ip_address(BA, "192.168.178.42"))
+
+    def test_is_valid_ipaddress_invalid(self):
+        self.assertFalse(BA.BaseAttack.is_valid_ip_address(BA, "192.168.1789.42"))
+
+    def test_ip_src_dst_equal_check_equal(self):
+        with self.assertRaises(SystemExit):
+            BA.BaseAttack.ip_src_dst_equal_check(BA, "192.168.178.42", "192.168.178.42")
+
+    def test_ip_src_dst_equal_check_unequal(self):
+        BA.BaseAttack.ip_src_dst_equal_check(BA, "192.168.178.42", "192.168.178.43")
+
+    def test_clean_whitespaces(self):
+        self.assertEqual("a\nb\rc\td\'e", BA.BaseAttack.clean_white_spaces(BA, "a\\nb\\rc\\td\\\'e"))
+
+    def test_generate_random_ipv4_address(self):
+        ip_list = BA.BaseAttack.generate_random_ipv4_address("Unknown", 10)
+        for ip in ip_list:
+            with self.subTest(ip=ip):
+                self.assertTrue(BA.BaseAttack._is_ip_address(ip))
+
+    def test_generate_random_ipv6_address(self):
+        ip_list = BA.BaseAttack.generate_random_ipv6_address(10)
+        for ip in ip_list:
+            with self.subTest(ip=ip):
+                self.assertTrue(BA.BaseAttack._is_ip_address(ip))
+
+    def test_generate_random_mac_address(self):
+        mac_list = BA.BaseAttack.generate_random_mac_address(10)
+        for mac in mac_list:
+            with self.subTest(mac=mac):
+                self.assertTrue(BA.BaseAttack._is_mac_address(mac))

code/Test/test_DDoS.py

@@ -0,0 +1,34 @@
+import unittest.mock as mock
+
+import Test.ID2TAttackTest as Test
+import ID2TLib.TestLibrary as Lib
+
+sha_basic_ddos = '87c6c9cf4b496b84fecfd758c1d891ff06fe234dba2f421a5ab8bd7d6d9239a5'
+sha_num_attackers_ddos = 'cbbb9b55d03a0efde965bbb8c38f6ba8a9acbd605cb2f3ac22a6ed6e3958f8e9'
+sha_dest_mac_length_zero_ddos = 'acf1d108ab3d4e76636c6b58e08296126a74fcf3936377588376a79716fffd60'
+sha_mss_none_ddos = '87c6c9cf4b496b84fecfd758c1d891ff06fe234dba2f421a5ab8bd7d6d9239a5'
+
+# TODO: improve coverage
+
+
+class UnitTestDDoS(Test.ID2TAttackTest):
+
+    @mock.patch('ID2TLib.Utility.get_attacker_config', side_effect=Lib.get_attacker_config)
+    def test_ddos_basic(self, mock_get_attacker_config):
+        self.checksum_test([['DDoSAttack']],
+                           sha_basic_ddos)
+
+    @mock.patch('ID2TLib.Utility.get_attacker_config', side_effect=Lib.get_attacker_config)
+    def test_ddos_num_attackers(self, mock_get_attacker_config):
+        self.checksum_test([['DDoSAttack', 'attackers.count=5']],
+                           sha_num_attackers_ddos)
+
+    @mock.patch('ID2TLib.Utility.get_attacker_config', side_effect=Lib.get_attacker_config)
+    @mock.patch('ID2TLib.Statistics.Statistics.get_mac_address', return_value=[])
+    def test_ddos_dest_mac_length_zero(self, mock_dest_mac, mock_get_attacker_config):
+        self.checksum_test([['DDoSAttack']], sha_dest_mac_length_zero_ddos)
+
+    @mock.patch('ID2TLib.Utility.get_attacker_config', side_effect=Lib.get_attacker_config)
+    @mock.patch('ID2TLib.Statistics.Statistics.get_most_used_mss', return_value=None)
+    def test_ddos_mss_none(self, mock_mss, mock_get_attacker_config):
+        self.checksum_test([['DDoSAttack']], sha_mss_none_ddos)

code/Test/test_EternalBlue.py

@@ -0,0 +1,11 @@
+import Test.ID2TAttackTest as Test
+
+sha_default = 'c707492a0493efcf46a569c91fe77685286402ddfdff3c79e64157b3324dc9f6'
+
+# TODO: improve coverage
+
+
+class UnitTestEternalBlue(Test.ID2TAttackTest):
+
+    def test_eternal_blue_default(self):
+        self.checksum_test([['EternalBlueExploit']], sha_default)

code/Test/test_FTPWinaXeExploit.py

@@ -0,0 +1,52 @@
+import unittest.mock as mock
+
+import ID2TLib.TestLibrary as Lib
+import Test.ID2TAttackTest as Test
+
+sha_ftp_basic = 'ad9bc7b55c3b0365c0f02ae9b9b7aafdb43acbdd8c8c274d30cb286821e772cc'
+sha_ftp_mac = '388831100c907cfc6815bcc1869f30d937be29091dd8e54a734eb52f14a23f3c'
+sha_ftp_random_ip_src = 'b18c0f1d15f1afb239116e1ccec20b03716412eea58ca969f7d2ede1749409e3'
+sha_not_empty_custom_payload_empty_file = '41186fc804fb2a8fb3605be3246a5246be927e3187ea82bd2fbe2097643863a8'
+sha_empty_custom_payload_not_empty_file = 'b1f43c3147dd3684b1db4d7d370801f25de693b632b97a95b933a4d296094f31'
+sha_valid_ip = 'ad9bc7b55c3b0365c0f02ae9b9b7aafdb43acbdd8c8c274d30cb286821e772cc'
+
+# TODO: improve coverage
+
+
+class UnitTestFTPWinaXeExploit(Test.ID2TAttackTest):
+
+    @mock.patch('ID2TLib.Utility.get_rnd_bytes', side_effect=Lib.get_bytes)
+    @mock.patch('ID2TLib.Utility.get_rnd_x86_nop', side_effect=Lib.get_x86_nop)
+    def test_ftp_basic(self, mock_get_rnd_x86_nop, mock_get_rnd_bytes):
+        self.checksum_test([['FTPWinaXeExploit']], sha_ftp_basic)
+
+    @mock.patch('ID2TLib.Utility.get_rnd_bytes', side_effect=Lib.get_bytes)
+    @mock.patch('ID2TLib.Utility.get_rnd_x86_nop', side_effect=Lib.get_x86_nop)
+    @mock.patch('ID2TLib.Statistics.Statistics.get_mac_address')
+    def test_ftp_mac(self, mock_mac_address, mock_get_rnd_x86_nop, mock_get_rnd_bytes):
+        mock_mac_address.return_value = Lib.test_pcap_empty
+        self.checksum_test([['FTPWinaXeExploit']], sha_ftp_mac)
+
+    @mock.patch('ID2TLib.Utility.get_rnd_bytes', side_effect=Lib.get_bytes)
+    @mock.patch('ID2TLib.Utility.get_rnd_x86_nop', side_effect=Lib.get_x86_nop)
+    def test_ftp_random_ip_src(self, mock_get_rnd_x86_nop, mock_get_rnd_bytes):
+        self.checksum_test([['FTPWinaXeExploit', 'ip.src.shuffle=1']], sha_ftp_random_ip_src)
+
+    @mock.patch('ID2TLib.Utility.get_rnd_bytes', side_effect=Lib.get_bytes)
+    @mock.patch('ID2TLib.Utility.get_rnd_x86_nop', side_effect=Lib.get_x86_nop)
+    def test_ftp_not_empty_custom_payload_empty_file(self, mock_get_rnd_x86_nop, mock_get_rnd_bytes):
+        self.checksum_test([['FTPWinaXeExploit', 'custom.payload=1']], sha_not_empty_custom_payload_empty_file)
+
+    @mock.patch('ID2TLib.Utility.get_rnd_bytes', side_effect=Lib.get_bytes)
+    @mock.patch('ID2TLib.Utility.get_rnd_x86_nop', side_effect=Lib.get_x86_nop)
+    @mock.patch('ID2TLib.Utility.check_payload_len')
+    @mock.patch('ID2TLib.Utility.get_bytes_from_file', return_value=b'AAAAA')
+    def test_ftp_empty_custom_payload_not_empty_file(self, mock_bytes_from_file, mock_payload_len, mock_get_rnd_x86_nop,
+                                                     mock_get_rnd_bytes):
+        self.checksum_test([['FTPWinaXeExploit', 'custom.payload.file=1']], sha_empty_custom_payload_not_empty_file)
+
+    @mock.patch('ID2TLib.Utility.get_rnd_bytes', side_effect=Lib.get_bytes)
+    @mock.patch('ID2TLib.Utility.get_rnd_x86_nop', side_effect=Lib.get_x86_nop)
+    @mock.patch('Attack.BaseAttack.BaseAttack.is_valid_ip_address', return_values=[False, True])
+    def test_ftp_invalid_ip(self, mock_valid_ip_check, mock_get_rnd_x86_nop, mock_get_rnd_bytes):
+        self.checksum_test([['FTPWinaXeExploit']], sha_valid_ip)

code/Test/test_Joomla.py

@@ -0,0 +1,11 @@
+import Test.ID2TAttackTest as Test
+
+sha_default = 'a45bd543ae7416cdc5fd76c886f48990b43075753931683407686aac2cfbc111'
+
+# TODO: improve coverage
+
+
+class UnitTestJoomla(Test.ID2TAttackTest):
+
+    def test_joomla_default(self):
+        self.checksum_test([['JoomlaRegPrivExploit']], sha_default)

code/Test/test_PortscanAttack.py

@@ -0,0 +1,41 @@
+import unittest.mock as mock
+
+import Test.ID2TAttackTest as Test
+
+sha_portscan_default = '6af539fb9f9a28f84a5c337a07dbdc1a11885c5c6de8f9a682bd74b89edc5130'
+sha_portscan_reverse_ports = '1c03342b7b94fdd1c9903d07237bc5239ebb7bd77a3dd137c9c378fa216c5382'
+sha_portscan_shuffle_dst_ports = '40485e47766438425900b787c4cda4ad1b5cd0d233b80f38bd45b5a88b70a797'
+sha_portscan_shuffle_src_ports = '48578b45e18bdbdc0a9f3f4cec160ccb58839250348ec4d3ec44c1b15da248de'
+sha_portscan_mss_value_zero = '8d32476a89262b78118a68867fff1d45c81f8ffb4970201f9d5ee3dfd94ba58a'
+sha_portscan_ttl_value_zero = 'ff8cf15d8e59856e0c6e43d81fa40180ebf2127042f376217cc2a20e4f21726e'
+sha_portscan_win_value_zero = 'b2fcbf72190ac3bf12192d0d7ee8c09ef87adb0d94a2610615ca76d8b577bbfb'
+sha_portscan_ip_src_random = 'c3939f30a40fa6e2164cc91dc4a7e823ca409492d44508e3edfc9d24748af0e5'
+
+# TODO: improve coverage
+
+
+class UnitTestPortscanAttack(Test.ID2TAttackTest):
+
+    def test_portscan_default(self):
+        self.checksum_test([['PortscanAttack']], sha_portscan_default)
+
+    def test_portscan_reverse_ports(self):
+        self.checksum_test([['PortscanAttack', 'port.dst.order-desc=1']], sha_portscan_reverse_ports)
+
+    def test_portscan_shuffle_dst_ports(self):
+        self.checksum_test([['PortscanAttack', 'port.dst.shuffle=1']], sha_portscan_shuffle_dst_ports)
+
+    def test_portscan_shuffle_src_ports(self):
+        self.checksum_test([['PortscanAttack', 'port.src.shuffle=1']], sha_portscan_shuffle_src_ports)
+
+    @mock.patch('ID2TLib.Statistics.Statistics.get_mss_distribution', return_value='')
+    def test_portscan_mss_length_zero(self, mock_mss_dis):
+        self.checksum_test([['PortscanAttack']], sha_portscan_mss_value_zero)
+
+    @mock.patch('ID2TLib.Statistics.Statistics.get_ttl_distribution', return_value='')
+    def test_portscan_ttl_length_zero(self, mock_ttl_dis):
+        self.checksum_test([['PortscanAttack']], sha_portscan_ttl_value_zero)
+
+    @mock.patch('ID2TLib.Statistics.Statistics.get_win_distribution', return_value='')
+    def test_portscan_win_length_zero(self, mock_win_dis):
+        self.checksum_test([['PortscanAttack']], sha_portscan_win_value_zero)

code/Test/test_Queries.py

@@ -0,0 +1,228 @@
+import random
+import unittest
+
+import ID2TLib.Controller as Ctrl
+import ID2TLib.TestLibrary as Test
+
+# TODO: improve coverage
+
+controller = Ctrl.Controller(pcap_file_path=Test.test_pcap, do_extra_tests=False)
+controller.load_pcap_statistics(flag_write_file=False, flag_recalculate_stats=True, flag_print_statistics=False)
+
+file_information = [('Pcap file', Test.test_pcap),
+                    ('Packets', 1998, 'packets'), ('Capture length', '25.4294414520264', 'seconds'),
+                    ('Capture start', '1970-01-01 01:01:45.647675'), ('Capture end', '1970-01-01 01:08:10.102034')]
+
+file_statistics = [('Avg. packet rate', 78.57034301757812, 'packets/sec'), ('Avg. packet size', 0.0, 'kbytes'),
+                   ('Avg. packets sent', 90.0, 'packets'), ('Avg. bandwidth in', 9.5290, 'kbit/s'),
+                   ('Avg. bandwidth out', 9.5290, 'kbit/s')]
+
+ip_addresses = ["10.0.2.15", "104.83.103.45", "13.107.21.200", "131.253.61.100", "172.217.23.142",
+                "172.217.23.174", "192.168.33.254", "204.79.197.200", "23.51.123.27", "35.161.3.50",
+                "52.11.17.245", "52.34.37.177", "52.39.210.199", "52.41.250.141", "52.85.173.182",
+                "54.149.74.139", "54.187.98.195", "54.192.44.108", "54.192.44.177", "72.247.178.113",
+                "72.247.178.67", "93.184.220.29"]
+ports = [53, 80, 443, 49157, 49160, 49163, 49164, 49165, 49166, 49167, 49168, 49169, 49170, 49171, 49172, 49173, 49174,
+         49175, 49176, 49177, 49178, 49179, 49180, 49181, 49182, 49183, 49184, 49185, 49186, 49187, 49188, 49189, 49190,
+         49191, 49192, 49193, 49194, 49195, 49196, 49197, 49247, 49323, 49470, 49636, 49695, 49798, 49927, 49935, 49945,
+         50262, 50836, 50968, 51143, 51166, 51350, 51451, 51669, 51713, 52033, 52135, 52399, 52520, 52644, 52697, 52743,
+         52786, 52964, 52981, 53059, 53234, 53461, 53691, 53708, 53745, 53836, 54049, 54446, 54593, 54598, 54652, 54663,
+         54717, 54853, 54930, 55004, 55018, 55119, 55125, 55299, 55310, 55463, 55650, 55667, 55752, 55843, 55851, 56146,
+         56325, 56567, 56589, 56750, 57049, 57179, 57275, 57520, 57653, 57840, 57957, 57991, 58401, 58440, 58645, 58797,
+         58814, 58905, 58913, 58943, 59380, 59408, 59461, 59467, 59652, 59660, 59718, 59746, 59844, 60006, 60209, 60414,
+         60422, 60659, 60696, 60708, 60756, 60827, 60840, 61181, 61300, 61592, 61718, 61738, 61769, 61807, 62412, 62428,
+         62447, 62490, 62625, 62626, 62664, 63425, 64096, 64121, 64137, 64252, 64334, 64337, 64479, 64509, 64637, 64807,
+         64811, 65448, 65487]
+
+
+class TestQueries(unittest.TestCase):
+    def test_get_file_information(self):
+        self.assertEqual(controller.statistics.get_file_information(), file_information)
+
+    def test_get_general_file_statistics(self):
+        file_stats = controller.statistics.get_general_file_statistics()
+        file_stats[3] = ('Avg. bandwidth in', round(file_stats[3][1], 4), 'kbit/s')
+        file_stats[4] = ('Avg. bandwidth out', round(file_stats[4][1], 4), 'kbit/s')
+        self.assertEqual(file_stats, file_statistics)
+
+    def test_get_capture_duration(self):
+        self.assertEqual(controller.statistics.get_capture_duration(), '25.4294414520264')
+
+    def test_get_pcap_timestamp_start(self):
+        self.assertEqual(controller.statistics.get_pcap_timestamp_start(), '1970-01-01 01:01:45.647675')
+
+    def test_get_pcap_timestamp_end(self):
+        self.assertEqual(controller.statistics.get_pcap_timestamp_end(), '1970-01-01 01:08:10.102034')
+
+    def test_get_pps_sent_1(self):
+        self.assertEqual(controller.statistics.get_pps_sent(ip_address='72.247.178.67'), 0)
+
+    def test_get_pps_sent_2(self):
+        self.assertEqual(controller.statistics.get_pps_sent(ip_address='10.0.2.15'), 32)
+
+    def test_get_pps_received_1(self):
+        self.assertEqual(controller.statistics.get_pps_received(ip_address='72.247.178.67'), 0)
+
+    def test_get_pps_received_2(self):
+        self.assertEqual(controller.statistics.get_pps_received(ip_address='10.0.2.15'), 46)
+
+    def test_get_packet_count(self):
+        self.assertEqual(controller.statistics.get_packet_count(), 1998)
+
+    def test_get_most_used_ip_address(self):
+        self.assertEqual(controller.statistics.get_most_used_ip_address(), '10.0.2.15')
+
+    def test_get_ttl_distribution_1(self):
+        self.assertEqual(controller.statistics.get_ttl_distribution(ipAddress='72.247.178.67'), {64: 5})
+
+    def test_get_ttl_distribution_2(self):
+        self.assertEqual(controller.statistics.get_ttl_distribution(ipAddress='10.0.2.15'), {128: 817})
+
+    def test_get_mss_distribution_1(self):
+        self.assertEqual(controller.statistics.get_mss_distribution(ipAddress='72.247.178.67'), {1460: 1})
+
+    def test_get_mss_distribution_2(self):
+        self.assertEqual(controller.statistics.get_mss_distribution(ipAddress='10.0.2.15'), {1460: 36})
+
+    def test_get_win_distribution_1(self):
+        self.assertEqual(controller.statistics.get_win_distribution(ipAddress='72.247.178.67'), {65535: 5})
+
+    def test_get_tos_distribution_1(self):
+        self.assertEqual(controller.statistics.get_tos_distribution(ipAddress='72.247.178.67'), {0: 5})
+
+    def test_get_tos_distribution_2(self):
+        self.assertEqual(controller.statistics.get_tos_distribution(ipAddress='10.0.2.15'), {0: 817})
+
+    def test_get_ip_address_count(self):
+        self.assertEqual(controller.statistics.get_ip_address_count(), 22)
+
+    def test_get_ip_addresses(self):
+        self.assertEqual(controller.statistics.get_ip_addresses(), ip_addresses)
+
+    def test_get_random_ip_address(self):
+        random.seed(5)
+        self.assertEqual(controller.statistics.get_random_ip_address(), '72.247.178.113')
+
+    def test_get_random_ip_address_count_2(self):
+        random.seed(5)
+        self.assertEqual(controller.statistics.get_random_ip_address(2), ['72.247.178.113', '23.51.123.27'])
+
+    def test_get_mac_address_1(self):
+        self.assertEqual(controller.statistics.get_mac_address(ipAddress='72.247.178.67'), '52:54:00:12:35:02')
+
+    def test_get_mac_address_2(self):
+        self.assertEqual(controller.statistics.get_mac_address(ipAddress='10.0.2.15'), '08:00:27:a3:83:43')
+
+    def test_get_most_used_mss(self):
+        self.assertEqual(controller.statistics.get_most_used_mss(ipAddress='10.0.2.15'), 1460)
+
+    def test_get_most_used_ttl(self):
+        self.assertEqual(controller.statistics.get_most_used_ttl(ipAddress='10.0.2.15'), 128)
+
+    def test_is_query_no_string(self):
+        self.assertFalse(controller.statistics.is_query(42))
+
+    def test_is_query_named_query(self):
+        self.assertTrue(controller.statistics.is_query('least_used(ipaddress)'))
+
+    def test_is_query_standard_query(self):
+        self.assertTrue(controller.statistics.is_query('SELECT * from ip_statistics'))
+
+    def test_calculate_standard_deviation(self):
+        self.assertEqual(controller.statistics.calculate_standard_deviation([1, 1, 2, 3, 5, 8, 13, 21]),
+                         6.609652033201143)
+
+    def test_calculate_entropy(self):
+        self.assertEqual(controller.statistics.calculate_entropy([1, 1, 2, 3, 5, 8, 13, 21]), 2.371389165297016)
+
+    def test_calculate_entropy_normalized(self):
+        self.assertEqual(controller.statistics.calculate_entropy([1, 1, 2, 3, 5, 8, 13, 21], normalized=True),
+                         (2.371389165297016, 0.7904630550990053))
+
+    def test_calculate_complement_packet_rates_1(self):
+        cpr = controller.statistics.calculate_complement_packet_rates(0)[0:9]
+        self.assertEqual(cpr, [(186.418564, 0), (186.418824, 0), (186.419346, 0), (186.445361, 0),
+                               (186.46954399999998, 0), (186.476234, 0), (186.477304, 0), (186.48606999999998, 0),
+                               (186.486761, 0)])
+
+    def test_calculate_complement_packet_rates_2(self):
+        cpr = controller.statistics.calculate_complement_packet_rates(42)[0:9]
+        self.assertEqual(cpr, [(186.418564, 41), (186.418824, 42), (186.419346, 42), (186.445361, 42),
+                               (186.46954399999998, 42), (186.476234, 42), (186.477304, 42), (186.48606999999998, 42),
+                               (186.486761, 42)])
+
+    # NAMED QUERY TESTS
+    def test_most_used_ipaddress(self):
+        self.assertEqual(controller.statistics.process_db_query('most_used(ipaddress)'), '10.0.2.15')
+
+    def test_most_used_macaddress(self):
+        self.assertEqual(controller.statistics.process_db_query('most_used(macaddress)'), '52:54:00:12:35:02')
+
+    def test_most_used_portnumber(self):
+        self.assertEqual(controller.statistics.process_db_query('most_used(portnumber)'), 443)
+
+    def test_most_used_protocolname(self):
+        self.assertEqual(controller.statistics.process_db_query('most_used(protocolname)'), 'IPv4')
+
+    def test_most_used_ttlvalue(self):
+        self.assertEqual(controller.statistics.process_db_query('most_used(ttlvalue)'), 64)
+
+    def test_most_used_mssvalue(self):
+        self.assertEqual(controller.statistics.process_db_query('most_used(mssvalue)'), 1460)
+
+    def test_most_used_winsize(self):
+        self.assertEqual(controller.statistics.process_db_query('most_used(winsize)'), 65535)
+
+    def test_most_used_ipclass(self):
+        self.assertEqual(controller.statistics.process_db_query('most_used(ipclass)'), 'A')
+
+    def test_least_used_ipaddress(self):
+        self.assertEqual(controller.statistics.process_db_query('least_used(ipaddress)'), '72.247.178.113')
+
+    def test_least_used_macaddress(self):
+        self.assertEqual(controller.statistics.process_db_query('least_used(macaddress)'), '08:00:27:a3:83:43')
+
+    def test_least_used_portnumber(self):
+        self.assertEqual(controller.statistics.process_db_query('least_used(portnumber)'), [58645, 59844])
+
+    def test_least_used_protocolname(self):
+        self.assertEqual(controller.statistics.process_db_query('least_used(protocolname)'), 'UDP')
+
+    def test_least_used_ttlvalue(self):
+        self.assertEqual(controller.statistics.process_db_query('least_used(ttlvalue)'), 255)
+
+    def test_avg_pktsreceived(self):
+        self.assertEqual(controller.statistics.process_db_query('avg(pktsreceived)'), 90.36363636363636)
+
+    def test_avg_pktssent(self):
+        self.assertEqual(controller.statistics.process_db_query('avg(pktssent)'), 90.36363636363636)
+
+    def test_avg_kbytesreceived(self):
+        self.assertEqual(controller.statistics.process_db_query('avg(kbytesreceived)'), 30.289683948863637)
+
+    def test_avg_kbytessent(self):
+        self.assertEqual(controller.statistics.process_db_query('avg(kbytessent)'), 30.289683948863637)
+
+    def test_avg_ttlvalue(self):
+        self.assertEqual(controller.statistics.process_db_query('avg(ttlvalue)'), 75.08695652173913)
+
+    def test_avg_mss(self):
+        self.assertEqual(controller.statistics.process_db_query('avg(mss)'), 1460.0)
+
+    def test_all_ipaddress(self):
+        self.assertEqual(controller.statistics.process_db_query('all(ipaddress)'), ip_addresses)
+
+    def test_all_ttlvalue(self):
+        self.assertEqual(controller.statistics.process_db_query('all(ttlvalue)'), [64, 128, 255])
+
+    def test_all_mss(self):
+        self.assertEqual(controller.statistics.process_db_query('all(mss)'), 1460)
+
+    def test_all_macaddress(self):
+        self.assertEqual(controller.statistics.process_db_query('all(macaddress)'), ['08:00:27:a3:83:43',
+                                                                                     '52:54:00:12:35:02'])
+    def test_all_portnumber(self):
+        self.assertEqual(controller.statistics.process_db_query('all(portnumber)'), ports)
+
+    def test_all_protocolname(self):
+        self.assertEqual(controller.statistics.process_db_query('all(protocolname)'), ['IPv4', 'TCP', 'UDP'])

code/Test/test_SMBLib.py

@@ -0,0 +1,58 @@
+import unittest
+
+import ID2TLib.SMBLib as SMBLib
+import ID2TLib.Utility as Utility
+
+
+class TestSMBLib(unittest.TestCase):
+
+    def test_get_smb_version_all(self):
+
+        for platform in Utility.platforms:
+            with self.subTest(platform):
+                result = SMBLib.get_smb_version(platform)
+                self.assertTrue((result in SMBLib.smb_versions_per_win.values() or
+                                 result in SMBLib.smb_versions_per_samba.values()))
+
+    def test_get_smb_version_invalid(self):
+
+        with self.assertRaises(SystemExit):
+            SMBLib.get_smb_version("abc")
+
+    def test_get_smb_version_mac(self):
+        self.assertEqual(SMBLib.get_smb_version("macos"), "2.1")
+
+    def test_get_smb_version_win(self):
+
+        win_platforms = {'win7', 'win10', 'winxp', 'win8.1', 'win8', 'winvista', 'winnt', "win2000"}
+
+        for platform in win_platforms:
+            with self.subTest(platform):
+                self.assertIn(SMBLib.get_smb_version(platform), SMBLib.smb_versions_per_win.values())
+
+    def test_get_smb_version_linux(self):
+        self.assertIn(SMBLib.get_smb_version("linux"), SMBLib.smb_versions_per_samba.values())
+
+    def test_get_smb_platform_data_invalid(self):
+
+        with self.assertRaises(SystemExit):
+            SMBLib.get_smb_platform_data("abc", 0)
+
+    def test_get_smb_platform_data_linux(self):
+        self.assertEqual((SMBLib.get_smb_platform_data("linux", 0)),
+                         ("ubuntu", SMBLib.security_blob_ubuntu, 0x5, 0x800000, 0))
+
+    def test_get_smb_platform_data_mac(self):
+        guid, blob, cap, d_size, time = SMBLib.get_smb_platform_data("macos", 0)
+        self.assertEqual((blob, cap, d_size, time), (SMBLib.security_blob_macos, 0x6, 0x400000, 0))
+        self.assertTrue(isinstance(guid, str) and len(guid) > 0)
+
+    def test_get_smb_platform_data_win(self):
+        guid, blob, cap, d_size, time = SMBLib.get_smb_platform_data("win7", 100)
+        self.assertEqual((blob, cap, d_size), (SMBLib.security_blob_windows, 0x7, 0x100000))
+        self.assertTrue(isinstance(guid, str) and len(guid) > 0)
+        self.assertTrue(time <= Utility.get_filetime_format(100))
+
+    def test_invalid_smb_version(self):
+        with self.assertRaises(SystemExit):
+            SMBLib.invalid_smb_version("abc")

code/Test/test_SMBLoris.py

@@ -0,0 +1,30 @@
+import ID2TLib.TestLibrary as Lib
+import Test.ID2TAttackTest as Test
+
+sha_default = 'cbfb154a80546ebcf0a0d5128bcc42e4d69228c1d97ea4dda49ba156703b78c2'
+sha_one_attacker = 'a316ba1a667318ef4b8d1bf5ffee3f58dfcd0221b0cc3ab62dd967379217eb27'
+sha_sixteen_attackers = '08b17b360ee9be1657e7c437e5aef354dac374ceca3b4ee437c45c0d9d03a2ef'
+sha_ips_in_pcap = 'f299e4139780869d9f02c25ba00f1cad483a4f215d6aef4079b93f7f7e1de22a'
+
+# TODO: improve coverage
+
+
+class UnitTestSMBLoris(Test.ID2TAttackTest):
+
+    def test_smbloris_default(self):
+        self.checksum_test([['SMBLorisAttack']], sha_default)
+
+    def test_smbloris_one_attacker(self):
+        self.checksum_test([['SMBLorisAttack', 'ip.src=192.168.1.240', 'ip.dst=192.168.1.210']], sha_one_attacker)
+
+    def test_smbloris_ips_in_pcap(self):
+        ip_src = 'ip.src='+Lib.test_pcap_ips[0]
+        ip_dst = 'ip.dst='+Lib.test_pcap_ips[1]
+        self.checksum_test([['SMBLorisAttack', ip_src, ip_dst]], sha_ips_in_pcap)
+
+    def test_smbloris_sixteen_attackers(self):
+        self.checksum_test([['SMBLorisAttack', 'ip.dst=192.168.1.210', 'attackers.count=16']], sha_sixteen_attackers)
+
+    def test_smbloris_same_ip_src_dst(self):
+        with self.assertRaises(SystemExit):
+            self.checksum_test([['SMBLorisAttack', 'ip.src=192.168.1.240', 'ip.dst=192.168.1.240']], sha_default)

code/Test/test_SMBScan.py

@@ -0,0 +1,70 @@
+import unittest.mock as mock
+
+import Test.ID2TAttackTest as Test
+
+sha_default = '213e194da7bc952cc093868c7450901b0fb93c7255d694eb37ea0b9b48bca65d'
+sha_one_victim_linux = '4928d421caaec8f2c4e5c5bb835b5521b705478779cbc8f343b77143a5a66995'
+sha_victim_range_winxp_hosting = '4c6cb5cb4f838e75b41af4feb2fd9a6fe7e1b226a38b3e8759ce3d31e5a2535e'
+sha_multiple_victims_macos = '0be79b9ad7346562f392e07a5156de978e02f4f25ae8d409b81cc6e0d726012c'
+sha_port_shuffle = '8ef501fa31135b8fea845a2be6a9605e0c3f9c4895b717f9206d485a669c2a73'
+sha_dest_mac_only = '0814dadb666e0056ef5b3a572a4971f333376b61e602acb84cb99c851845f016'
+sha_ip_src_shuffle = '6c0c9ccbedb631e4965ec36932276a1bd73b8a4aca5a5c46f01fd0a2800a064f'
+sha_smb2 = '8755a901295a90362d8041ecf1243a31fff582f5fe64555205625263c253476e'
+
+# TODO: improve coverage
+
+
+class UnitTestSMBScan(Test.ID2TAttackTest):
+
+    def test_smbscan_default(self):
+        with mock.patch("ID2TLib.Utility.get_rnd_os", return_value="win7"):
+            self.checksum_test([['SMBScanAttack']], sha_default)
+
+    def test_smbscan_one_victim_linux(self):
+        with mock.patch("ID2TLib.Utility.get_rnd_os", return_value="linux"):
+            self.checksum_test([['SMBScanAttack', 'ip.src=192.168.178.1', 'ip.dst=192.168.178.10']],
+                               sha_one_victim_linux)
+
+    def test_smbscan_victim_range_winxp_hosting(self):
+        with mock.patch("ID2TLib.Utility.get_rnd_os", return_value="winxp"):
+            self.checksum_test([['SMBScanAttack', 'ip.src=192.168.178.1', 'ip.dst=192.168.178.5',
+                                'ip.dst.end=192.168.178.10', 'hosting.ip=192.168.178.5']],
+                               sha_victim_range_winxp_hosting)
+
+    def test_smbscan_multiple_victims_macos(self):
+        with mock.patch("ID2TLib.Utility.get_rnd_os", return_value="macos"):
+            self.checksum_test([['SMBScanAttack', 'ip.src=192.168.178.1',
+                                'ip.dst=192.168.178.10,192.168.178.15,192.168.178.20',
+                                'hosting.ip=192.168.178.15,192.168.178.20']], sha_multiple_victims_macos)
+
+    def test_smbscan_invalid_smb_version(self):
+        with self.assertRaises(SystemExit):
+            self.checksum_test([['SMBScanAttack', 'protocol.version=42']], 'somehash')
+
+    def test_smbscan_invalid_smb_platform(self):
+        with self.assertRaises(SystemExit):
+            self.checksum_test([['SMBScanAttack', 'hosting.version=1337']], 'somehash')
+
+    def test_smbscan_port_shuffle(self):
+        with mock.patch("ID2TLib.Utility.get_rnd_os", return_value="win7"):
+            self.checksum_test([['SMBScanAttack', 'ip.src=192.168.178.1', 'ip.dst=192.168.178.5',
+                                'ip.dst.end=192.168.178.10', 'hosting.ip=192.168.178.5', 'port.src.shuffle=false']],
+                               sha_port_shuffle)
+
+    def test_smbscan_dest_mac_only(self):
+        with mock.patch("ID2TLib.Utility.get_rnd_os", return_value="win7"):
+            self.checksum_test([['SMBScanAttack', 'ip.src=192.168.178.1',
+                                'mac.dst=00:0C:29:9C:70:64']], sha_dest_mac_only)
+
+    def test_smbscan_src_ip_shuffle(self):
+        with mock.patch("ID2TLib.Utility.get_rnd_os", return_value="win7"):
+            self.checksum_test([['SMBScanAttack', 'ip.src=192.168.178.1', 'ip.dst=192.168.178.5',
+                                'ip.dst.end=192.168.178.10', 'hosting.ip=192.168.178.5', 'ip.src.shuffle=True']],
+                               sha_ip_src_shuffle)
+
+    def test_smbscan_smb2(self):
+
+        with mock.patch("ID2TLib.Utility.get_rnd_os", return_value="linux"):
+            self.checksum_test([['SMBScanAttack', 'ip.src=192.168.178.1', 'ip.dst=192.168.178.5',
+                                'ip.dst.end=192.168.178.10', 'hosting.ip=192.168.178.5', 'protocol.version=2.1',
+                                'hosting.version=2.1']], sha_smb2)

code/Test/test_SQLi.py

@@ -0,0 +1,11 @@
+import Test.ID2TAttackTest as Test
+
+sha_default = 'a130ecdaf5fd8c09ef8418d2dbe7bd68c54e922553eb9fa703df016115393a46'
+
+# TODO: improve coverage
+
+
+class UnitTestSQLi(Test.ID2TAttackTest):
+
+    def test_sqli_default(self):
+        self.checksum_test([['SQLiAttack']], sha_default)

code/Test/test_Utility.py

@@ -0,0 +1,230 @@
+import unittest
+
+import ID2TLib.TestLibrary as Lib
+import ID2TLib.Utility as Utility
+
+# TODO: improve coverage
+
+
+class TestUtility(unittest.TestCase):
+
+    def test_update_timestamp_no_delay(self):
+        self.assertTrue(100+10/5 >= Utility.update_timestamp(100, 5) >= 100+1/5)
+
+    def test_update_timestamp_with_delay(self):
+        self.assertTrue(100+1/5+10*100 >= Utility.update_timestamp(100, 5, 10) >= 100+1/5+10)
+
+    def test_update_timestamp_comparison(self):
+        self.assertTrue(Utility.update_timestamp(100, 5) <= Utility.update_timestamp(100, 5, 10))
+
+    def test_get_interval_pps_below_max(self):
+        cipps = [(5, 1), (10, 2), (15, 3)]
+        self.assertEqual(Utility.get_interval_pps(cipps, 3), 1)
+        self.assertEqual(Utility.get_interval_pps(cipps, 7), 2)
+        self.assertEqual(Utility.get_interval_pps(cipps, 12), 3)
+
+    def test_get_interval_pps_above_max(self):
+        cipps = [(5, 1), (10, 2), (15, 3)]
+        self.assertEqual(Utility.get_interval_pps(cipps, 30), 3)
+
+    # Errors if empty list and result bad if only one list
+    def test_get_nth_random_element_equal_no(self):
+        letters = ["A", "B", "C"]
+        numbers = [1, 2, 3]
+        results = [("A", 1), ("B", 2), ("C", 3)]
+        self.assertIn(Utility.get_nth_random_element(letters, numbers), results)
+
+    def test_get_nth_random_element_unequal_no(self):
+        letters = ["A", "B", "C"]
+        numbers = [1, 2]
+        results = [("A", 1), ("B", 2)]
+        self.assertIn(Utility.get_nth_random_element(letters, numbers), results)
+
+    # TODO: ???
+    #def test_get_nth_random_element_single_list(self):
+        #letters = ["A", "B", "C"]
+        #self.assertIn(Utility.get_nth_random_element(letters), letters)
+
+    def test_index_increment_not_max(self):
+        self.assertEqual(Utility.index_increment(5, 10), 6)
+
+    def test_index_increment_max(self):
+        self.assertEqual(Utility.index_increment(10, 10), 0)
+
+    # Correct?
+    def test_index_increment_max2(self):
+        self.assertEqual(Utility.index_increment(9, 10), 0)
+
+    def test_get_rnd_os(self):
+        self.assertIn(Utility.get_rnd_os(), Utility.platforms)
+
+    def test_check_platform_valid(self):
+        try:
+            Utility.check_platform("linux")
+        except SystemExit:
+            self.fail()
+
+    def test_check_platform_invalid(self):
+        with self.assertRaises(SystemExit):
+            Utility.check_platform("abc")
+
+    def test_get_ip_range_forwards(self):
+        start = "192.168.178.254"
+        end = "192.168.179.1"
+        result = ["192.168.178.254", "192.168.178.255", "192.168.179.0", "192.168.179.1"]
+        self.assertEqual(Utility.get_ip_range(start, end), result)
+
+    def test_get_ip_range_backwards(self):
+        end = "192.168.178.254"
+        start = "192.168.179.1"
+        result = ["192.168.179.1", "192.168.179.0", "192.168.178.255", "192.168.178.254"]
+        self.assertEqual(Utility.get_ip_range(start, end), result)
+
+    def test_get_ip_range_equal(self):
+        end = "192.168.178.254"
+        start = "192.168.178.254"
+        result = ["192.168.178.254"]
+        self.assertEqual(Utility.get_ip_range(start, end), result)
+
+    def test_generate_source_port_from_platform_invalid(self):
+        with self.assertRaises(SystemExit):
+            Utility.generate_source_port_from_platform("abc")
+
+    def test_generate_source_port_from_platform_oldwin_firstport(self):
+        self.assertTrue(1024 <= Utility.generate_source_port_from_platform("winxp") <= 5000)
+
+    def test_generate_source_port_from_platform_oldwin_nextport(self):
+        self.assertEqual(Utility.generate_source_port_from_platform("winxp", 2000), 2001)
+
+    def test_generate_source_port_from_platform_oldwin_maxport(self):
+        self.assertTrue(1024 <= Utility.generate_source_port_from_platform("winxp", 5000) <= 5000)
+
+    def test_generate_source_port_from_platform_linux(self):
+        self.assertTrue(32768 <= Utility.generate_source_port_from_platform("linux") <= 61000)
+
+    def test_generate_source_port_from_platform_newwinmac_firstport(self):
+        self.assertTrue(49152 <= Utility.generate_source_port_from_platform("win7") <= 65535)
+
+    def test_generate_source_port_from_platform_newwinmac_nextport(self):
+        self.assertEqual(Utility.generate_source_port_from_platform("win7", 50000), 50001)
+
+    def test_generate_source_port_from_platform_newwinmac_maxport(self):
+        self.assertTrue(49152 <= Utility.generate_source_port_from_platform("win7", 65535) <= 65535)
+
+    # TODO: get_filetime_format Test
+
+    def test_get_rnd_boot_time_invalid(self):
+        with self.assertRaises(SystemExit):
+            Utility.get_rnd_boot_time(10, "abc")
+
+    def test_get_rnd_boot_time_linux(self):
+        self.assertTrue(Utility.get_rnd_boot_time(100, "linux") < 100)
+
+    def test_get_rnd_boot_time_macos(self):
+        self.assertTrue(Utility.get_rnd_boot_time(100, "macos") < 100)
+
+    def test_get_rnd_boot_time_win(self):
+        self.assertTrue(Utility.get_rnd_boot_time(100, "win7") < 100)
+
+    def test_get_rnd_x86_nop_len(self):
+        result = Utility.get_rnd_x86_nop(1000)
+        self.assertEqual(len(result), 1000)
+
+    def test_get_rnd_x86_nop_with_sideeffects(self):
+        result = Utility.get_rnd_x86_nop(1000, False)
+        correct = True
+        for byte in result:
+            if byte.to_bytes(1, "little") not in Utility.x86_nops and byte.to_bytes(1, "little") not in Utility.x86_pseudo_nops:
+                correct = False
+        self.assertTrue(correct)
+
+    def test_get_rnd_x86_nop_without_sideeffects(self):
+        result = Utility.get_rnd_x86_nop(1000, True)
+        correct = True
+        for byte in result:
+            if byte.to_bytes(1, "little") in Utility.x86_pseudo_nops:
+                correct = False
+        self.assertTrue(correct)
+
+    def test_get_rnd_x86_nop_filter(self):
+        result = Utility.get_rnd_x86_nop(1000, False, Utility.x86_nops.copy())
+        correct = True
+        for byte in result:
+            if byte.to_bytes(1, "little") in Utility.x86_nops:
+                correct = False
+        self.assertTrue(correct)
+
+    def test_get_rnd_x86_nop_single_filter(self):
+        result = Utility.get_rnd_x86_nop(1000, False, b'\x20')
+        correct = True
+        for byte in result:
+            if byte.to_bytes(1, "little") == b'\x20':
+                correct = False
+        self.assertTrue(correct)
+
+    def test_get_rnd_bytes_number(self):
+        result = Utility.get_rnd_bytes(1000)
+        self.assertEqual(len(result), 1000)
+
+    def test_get_rnd_bytes_filter(self):
+        result = Utility.get_rnd_bytes(1000, Utility.x86_pseudo_nops.copy())
+        correct = True
+        for byte in result:
+            if byte.to_bytes(1, "little") in Utility.x86_pseudo_nops:
+                correct = False
+        self.assertTrue(correct)
+
+    def test_get_bytes_from_file_invalid_path(self):
+        with self.assertRaises(SystemExit):
+            Utility.get_bytes_from_file(Lib.test_resource_dir + "/NonExistingFile.txt")
+
+    def test_get_bytes_from_file_invalid_header(self):
+        with self.assertRaises(SystemExit):
+            Utility.get_bytes_from_file(Lib.test_resource_dir + "/InvalidHeader.txt")
+
+    def test_get_bytes_from_file_invalid_hexfile(self):
+        with self.assertRaises(SystemExit):
+            Utility.get_bytes_from_file(Lib.test_resource_dir + "/InvalidHexFile.txt")
+
+    def test_get_bytes_from_file_invalid_strfile(self):
+        with self.assertRaises(SystemExit):
+            Utility.get_bytes_from_file(Lib.test_resource_dir + "/InvalidStringFile.txt")
+
+    def test_get_bytes_from_file_str(self):
+        result = Utility.get_bytes_from_file(Lib.test_resource_dir + "/StringTestFile.txt")
+        self.assertEqual(result, b'This is a string-test')
+
+    def test_get_bytes_from_file_hex(self):
+        result = Utility.get_bytes_from_file(Lib.test_resource_dir + "/HexTestFile.txt")
+        self.assertEqual(result, b'\xab\xcd\xef\xff\x10\xff\xaa\xab')
+
+    def test_handle_most_used_outputs_empty(self):
+        self.assertIsNone(Utility.handle_most_used_outputs([]))
+
+    def test_handle_most_used_outputs_one(self):
+        test_input = "SomeTest"
+        self.assertEqual(Utility.handle_most_used_outputs(test_input), test_input)
+
+    def test_handle_most_used_outputs_one_list(self):
+        test_input = ["SomeTest"]
+        self.assertEqual(Utility.handle_most_used_outputs(test_input), test_input[0])
+
+    def test_handle_most_used_outputs_list_sorted(self):
+        test_input = [0, 1, 2, 3, 4]
+        self.assertEqual(Utility.handle_most_used_outputs(test_input), 0)
+
+    def test_handle_most_used_outputs_list_unsorted(self):
+        test_input = [2, 4, 0, 1, 3]
+        self.assertEqual(Utility.handle_most_used_outputs(test_input), 0)
+
+    def test_check_payload_len_exceeded(self):
+        with self.assertRaises(SystemExit):
+            Utility.check_payload_len(10, 5)
+
+    def test_check_payload_len_valid(self):
+        try:
+            Utility.check_payload_len(5, 10)
+        except SystemExit:
+            self.fail()
+
+    # TODO: get_attacker_config Tests

resources/install_dependencies.sh

@@ -65,7 +65,7 @@ install_pkg_darwin()
 
 install_pip()
 {
-    PYTHON_MODULES="lea numpy matplotlib scapy-python3 scipy"
+    PYTHON_MODULES="lea numpy matplotlib scapy-python3 scipy coverage"
     echo -e "Python modules: Checking..."
 
     # Check first to avoid unnecessary sudo

resources/test/HexTestFile.txt

@@ -0,0 +1,2 @@
+hex
+"abcd ef \xff10\ff 'xaa' x \ ab"

resources/test/InvalidHeader.txt

@@ -0,0 +1,2 @@
+InvalidHeader
+The header above is invalid because it is not 'hex' or 'str'

resources/test/InvalidHexFile.txt

@@ -0,0 +1,2 @@
+hex
+This is not a valid hexdump

resources/test/InvalidStringFile.txt

@@ -0,0 +1,3 @@
+str
+This is a
+invalid string-test

resources/test/StringTestFile.txt

@@ -0,0 +1,2 @@
+str
+This is a string-test