Merge branch 'unittest_order' into merge_test

code/Test/ID2TAttackTest.py

@@ -3,6 +3,7 @@ import inspect
 
 import ID2TLib.Controller as Ctrl
 import ID2TLib.TestLibrary as Lib
+import scapy.utils as pcr
 
 
 class ID2TAttackTest(unittest.TestCase):
@@ -44,3 +45,63 @@ class ID2TAttackTest(unittest.TestCase):
             Lib.clean_up(controller)
         else:
             Lib.rename_test_result_files(controller, caller_function, attack_sub_dir, test_sub_dir)
+
+    def order_test(self, attack_args, seed=None, 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):
+        """
+        Checks if the result of an attack includes all packets in correct order.
+
+        :param attack_args: A list of attacks with their attack parameters (as defined in Controller.process_attacks).
+        :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:
+            path = controller.pcap_dest_path
+            file = pcr.RawPcapReader(path)
+            packet_a = file.read_packet()
+            packet_b = file.read_packet()
+            i = 0
+
+            while packet_b is not None:
+
+                time_a = packet_a[2][0:2]
+                time_b = packet_b[2][0:2]
+
+                if time_a[0] > time_b[0]:
+                    file.close()
+                    self.fail("Packet order incorrect at: " + str(i+1) + "-" + str(i+2) +
+                              ". Current time: " + str(time_a) + " Next time: " + str(time_b))
+                elif time_a[0] == time_b[0]:
+                    if time_a[1] > time_b[1]:
+                        file.close()
+                        self.fail("Packet order incorrect at: " + str(i + 1) + "-" + str(i + 2) +
+                                  ". Current time: " + str(time_a) + " Next time: " + str(time_b))
+
+                packet_a = packet_b
+                packet_b = file.read_packet()
+                i += 1
+
+            file.close()
+
+        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_DDoS.py

@@ -32,3 +32,6 @@ class UnitTestDDoS(Test.ID2TAttackTest):
     @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)
+
+    def test_ddos_order(self):
+        self.order_test([['DDoSAttack', 'attackers.count=2']])

code/Test/test_EternalBlue.py

@@ -22,3 +22,5 @@ class UnitTestEternalBlue(Test.ID2TAttackTest):
         self.checksum_test([['EternalBlueExploit', ip_src, ip_dst, 'mac.src=00:0C:21:1C:60:61',
                              'mac.dst=04:0C:32:2C:63:62', 'port.src=1337', 'port.dst=42']], sha_multiple_params)
 
+    def test_eternalblue_order(self):
+        self.order_test([['EternalBlueExploit']])

code/Test/test_FTPWinaXeExploit.py

@@ -50,3 +50,6 @@ class UnitTestFTPWinaXeExploit(Test.ID2TAttackTest):
     @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)
+
+    def test_ftp_order(self):
+        self.order_test([['FTPWinaXeExploit']])

code/Test/test_Joomla.py

@@ -22,3 +22,6 @@ class UnitTestJoomla(Test.ID2TAttackTest):
         self.checksum_test([['JoomlaRegPrivExploit', ip_src, ip_dst, 'mac.src=00:0C:21:1C:60:61',
                              'mac.dst=04:0C:32:2C:63:62', 'port.dst=42',
                              'target.host=www.ihopethisisnotarealwebsite.com']], sha_multiple_params)
+
+    def test_joomla_order(self):
+        self.order_test([['JoomlaRegPrivExploit']])

code/Test/test_PortscanAttack.py

@@ -26,3 +26,6 @@ class UnitTestPortscanAttack(Test.ID2TAttackTest):
 
     def test_portscan_ips_not_in_pcap(self):
         self.checksum_test([['PortscanAttack', 'ip.src=1.1.1.1', 'ip.dst=2.2.2.2']], sha_portscan_ips_not_in_pcap)
+
+    def test_portscan_order(self):
+        self.order_test([['PortscanAttack']])

code/Test/test_SMBLoris.py

@@ -28,3 +28,6 @@ class UnitTestSMBLoris(Test.ID2TAttackTest):
     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)
+
+    def test_smbloris_order(self):
+        self.order_test([['SMBLorisAttack', 'attackers.count=1']])

code/Test/test_SMBScan.py

@@ -68,3 +68,6 @@ class UnitTestSMBScan(Test.ID2TAttackTest):
             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)
+
+    def test_smbscan_order(self):
+        self.order_test([['SMBScanAttack']])

code/Test/test_SQLi.py

@@ -22,3 +22,5 @@ class UnitTestSQLi(Test.ID2TAttackTest):
         self.checksum_test([['SQLiAttack', ip_src, ip_dst, 'mac.src=00:0C:21:1C:60:61',
                              'mac.dst=04:0C:32:2C:63:62', 'port.dst=42',
                              'target.host=www.ihopethisisnotarealwebsite.com']], sha_multiple_params)
+    def test_sqli_order(self):
+        self.order_test([['SQLiAttack']])