port-param, test-improvement and additional-mapping-fields

code/Attack/AttackParameters.py

@@ -50,6 +50,9 @@ class Parameter(Enum):
     IP_REUSE_EXTERNAL = 'ip.reuse.external'  # percentage of public IPs in original PCAP to be reused
     # recommended type: Positive Integer between 0 and 100 ------------------------------------
     PACKET_PADDING = 'packet.padding'
+    # calculate the destination port based on the hostname (like some botnets do)
+    # otherwise the destination port is a normal ephemeral port
+    BOTNET_DST_PORT_CALCULATION = "botnet.dstportcalculation"
 
 class ParameterTypes(Enum):
     """

code/Attack/MembersMgmtCommAttack.py

@@ -109,7 +109,8 @@ class MembersMgmtCommAttack(BaseAttack.BaseAttack):
             Param.NAT_PRESENT: ParameterTypes.TYPE_BOOLEAN,
 
             # the base PCAP for the TTL distribution
-            Param.TTL_FROM_CAIDA: ParameterTypes.TYPE_BOOLEAN
+            Param.TTL_FROM_CAIDA: ParameterTypes.TYPE_BOOLEAN,
+            Param.BOTNET_DST_PORT_CALCULATION: ParameterTypes.TYPE_BOOLEAN
         }
 
         # create dict with MessageType values for fast name lookup
@@ -154,6 +155,7 @@ class MembersMgmtCommAttack(BaseAttack.BaseAttack):
 
         # choose the input PCAP as default base for the TTL distribution
         self.add_param_value(Param.TTL_FROM_CAIDA, False)
+        self.add_param_value(Param.BOTNET_DST_PORT_CALCULATION, True)
 
 
     def generate_attack_pcap(self, context):
@@ -534,12 +536,13 @@ class MembersMgmtCommAttack(BaseAttack.BaseAttack):
 
         portSelector = PortSelectors.LINUX
         # create port configurations for the bots
+        calculate_dst_port = self.get_param_value(Param.BOTNET_DST_PORT_CALCULATION)
         for bot in bot_configs:
             bot_configs[bot]["SrcPort"] = portSelector.select_port_udp()
-            bot_configs[bot]["DstPort"] = Generator.gen_random_server_port()
-
-        # print(local_init_ids)
-        # print(bot_configs)
+            if calculate_dst_port:
+                bot_configs[bot]["DstPort"] = Generator.gen_random_server_port()
+            else:
+                bot_configs[bot]["DstPort"] = portSelector.select_port_udp()
 
         # assign realistic TTL for every bot
         if self.get_param_value(Param.TTL_FROM_CAIDA):

code/ID2TLib/Botnet/MessageMapping.py

@@ -1,5 +1,6 @@
 import os.path
 from xml.dom.minidom import *
+import datetime
 
 
 class MessageMapping:
@@ -33,7 +34,11 @@ class MessageMapping:
             mapping.setAttribute("Src", str(message.src["ID"]))
             mapping.setAttribute("Dst", str(message.dst["ID"]))
             mapping.setAttribute("Type", str(message.type.value))
+
+            dt = datetime.datetime.fromtimestamp(message.time)
             mapping.setAttribute("Time", str(message.time))
+            mapping.setAttribute("Time-Datetime", dt.strftime("%Y-%m-%d %H:%M:%S.") + str(dt.microsecond))
+            mapping.setAttribute("Time-Timeonly", dt.strftime("%H:%M:%S.") + str(dt.microsecond))
 
             packet = self.id_to_packet.get(message.msg_id)
             mapping.setAttribute(self.ATTR_HAS_PACKET, "true" if packet is not None else "false")

test/test_pcap_comparator.py

@@ -20,6 +20,17 @@ class PcapComparison(unittest.TestCase):
 
     OUTPUT_FILES_PREFIX_LINE = "Output files created:"
 
+    def __init__(self, *args, **kwargs):
+        unittest.TestCase.__init__(self, *args, **kwargs)
+
+        # params to call id2t with, as a list[list[str]]
+        # do a round of testing for each list[str] we get
+        # if none generate some params itself
+        self.id2t_params = None
+
+    def set_id2t_params(self, params: "list[list[str]]"):
+        self.id2t_params = params
+
     def setUp(self):
         self.generated_files = []
         self.keep_files = []
@@ -28,7 +39,14 @@ class PcapComparison(unittest.TestCase):
         input_pcap = os.environ.get(self.PCAP_ENVIRONMENT_VALUE, self.DEFAULT_PCAP)
         seed = os.environ.get(self.SEED_ENVIRONMENT_VALUE, self.DEFAULT_SEED)
 
-        command_args = [self.ID2T_LOCATION, "-i", input_pcap, "--seed", seed, "-a", "MembersMgmtCommAttack"]
+        if self.id2t_params is None:
+            self.id2t_params = self.random_id2t_params()
+
+        for params in self.id2t_params:
+            self.do_test_round(input_pcap, seed, params)
+
+    def do_test_round(self, input_pcap, seed, additional_params):
+        command_args = [self.ID2T_LOCATION, "-i", input_pcap, "--seed", seed, "-a", "MembersMgmtCommAttack"] + additional_params
         command = " ".join(map(shlex.quote, command_args))
         self.print_warning("The command that gets executed is:", command)
 
@@ -44,7 +62,8 @@ class PcapComparison(unittest.TestCase):
 
             pcap = self.find_pcap(files)
             if generated_pcap is not None:
-                try: self.compare_pcaps(generated_pcap, pcap)
+                try:
+                    self.compare_pcaps(generated_pcap, pcap)
                 except AssertionError as e:
                     self.keep_files = [generated_pcap, pcap]
                     raise e
@@ -52,7 +71,7 @@ class PcapComparison(unittest.TestCase):
                 generated_pcap = pcap
 
             self.print_warning()
-            time.sleep(1) # let some time pass between calls because files are based on the time
+            time.sleep(1)  # let some time pass between calls because files are based on the time
 
     def tearDown(self):
         self.print_warning("Cleaning up files generated by the test-calls...")
@@ -109,8 +128,22 @@ class PcapComparison(unittest.TestCase):
     def print_warning(self, *text):
         print(*text, file=sys.stderr)
 
+    def random_id2t_params(self):
+        param = lambda key, val: "-p%s=%s" % (str(key), str(val))
+
+        return [
+            []
+        ]
+
 if __name__ == "__main__":
+    import sys
+
+    # parameters for this program are interpreted as id2t-parameters
+    id2t_args = sys.argv[1:]
+    comparison = PcapComparison("test_determinism")
+    if id2t_args: comparison.set_id2t_params([id2t_args])
+
     suite = unittest.TestSuite()
-    suite.addTest(PcapComparison("test_determinism"))
+    suite.addTest(comparison)
 
     unittest.TextTestRunner().run(suite)