fixed packet features

src/cicflowmeter/constants.py

@@ -1,3 +1,4 @@
 EXPIRED_UPDATE = 240
 CLUMP_TIMEOUT = 0.001
 ACTIVE_TIMEOUT = 0.005
+BULK_BOUND = 4

src/cicflowmeter/features/flow_bytes.py

@@ -230,3 +230,43 @@ class FlowBytes:
         """
         feat = self.feature
         return [packet["IP"].ttl for packet, _ in feat.packets][0]
+
+    def get_bytes_per_bulk(self, packet_direction):
+        if packet_direction == PacketDirection.FORWARD:
+            if self.feature.forward_bulk_count != 0:
+                return self.feature.forward_bulk_size / self.feature.forward_bulk_count
+        else:
+            if self.feature.backward_bulk_count != 0:
+                return (
+                    self.feature.backward_bulk_size / self.feature.backward_bulk_count
+                )
+        return 0
+
+    def get_packets_per_bulk(self, packet_direction):
+        if packet_direction == PacketDirection.FORWARD:
+            if self.feature.forward_bulk_count != 0:
+                return (
+                    self.feature.forward_bulk_packet_count
+                    / self.feature.forward_bulk_count
+                )
+        else:
+            if self.feature.backward_bulk_count != 0:
+                return (
+                    self.feature.backward_bulk_packet_count
+                    / self.feature.backward_bulk_count
+                )
+        return 0
+
+    def get_bulk_rate(self, packet_direction):
+        if packet_direction == PacketDirection.FORWARD:
+            if self.feature.forward_bulk_count != 0:
+                return (
+                    self.feature.forward_bulk_size / self.feature.forward_bulk_duration
+                )
+        else:
+            if self.feature.backward_bulk_count != 0:
+                return (
+                    self.feature.backward_bulk_size
+                    / self.feature.backward_bulk_duration
+                )
+        return 0

src/cicflowmeter/features/packet_count.py

@@ -55,6 +55,14 @@ class PacketCount:
             return backward_size / forward_size
         return 0
 
+    @staticmethod
+    def get_payload(packet):
+        if "TCP" in packet:
+            return packet["TCP"].payload
+        elif "UDP" in packet:
+            return packet["UDP"].payload
+        return 0
+
     def has_payload(self, packet_direction=None) -> int:
         """Calculates download and upload ratio.
 
@@ -62,25 +70,19 @@ class PacketCount:
             int: packets
         """
 
-        def get_payload(packet):
-            if "TCP" in packet:
-                return packet["TCP"].payload
-            elif "UDP" in packet:
-                return packet["UDP"].payload
-            return 0
-
         if packet_direction is not None:
             return len(
                 [
                     packet
                     for packet, direction in self.feature.packets
-                    if direction == packet_direction and len(get_payload(packet)) > 0
+                    if direction == packet_direction
+                    and len(self.get_payload(packet)) > 0
                 ]
             )
         return len(
             [
                 packet
                 for packet, direction in self.feature.packets
-                if len(get_payload(packet)) > 0
+                if len(self.get_payload(packet)) > 0
             ]
         )

src/cicflowmeter/features/packet_time.py

@@ -79,7 +79,7 @@ class PacketTime:
 
         """
 
-        return 1e6 * (max(self._get_packet_times()) - min(self._get_packet_times()))
+        return max(self._get_packet_times()) - min(self._get_packet_times())
 
     def get_var(self):
         """Calculates the variation of packet times in a network flow.

src/cicflowmeter/flow.py

@@ -39,12 +39,28 @@ class Flow:
             PacketDirection.REVERSE: 0,
         }
 
-        self.subflows = 0
         self.start_active = 0
         self.last_active = 0
         self.active = []
         self.idle = []
 
+        self.forward_bulk_last_timestamp = 0
+        self.forward_bulk_start_tmp = 0
+        self.forward_bulk_count = 0
+        self.forward_bulk_count_tmp = 0
+        self.forward_bulk_duration = 0
+        self.forward_bulk_packet_count = 0
+        self.forward_bulk_size = 0
+        self.forward_bulk_size_tmp = 0
+        self.backward_bulk_last_timestamp = 0
+        self.backward_bulk_start_tmp = 0
+        self.backward_bulk_count = 0
+        self.backward_bulk_count_tmp = 0
+        self.backward_bulk_duration = 0
+        self.backward_bulk_packet_count = 0
+        self.backward_bulk_size = 0
+        self.backward_bulk_size_tmp = 0
+
     def get_data(self) -> dict:
         """This method obtains the values of the features extracted from each flow.
 
@@ -82,7 +98,7 @@ class Flow:
             "protocol": self.protocol,
             # Basic information from packet times
             "timestamp": packet_time.get_time_stamp(),
-            "flow_duration": packet_time.get_duration(),
+            "flow_duration": 1e6 * packet_time.get_duration(),
             "flow_byts_s": flow_bytes.get_rate(),
             "flow_pkts_s": packet_count.get_rate(),
             "fwd_pkts_s": packet_count.get_rate(PacketDirection.FORWARD),
@@ -150,6 +166,12 @@ class Flow:
             "idle_min": idle_stat["min"],
             "idle_mean": idle_stat["mean"],
             "idle_std": idle_stat["std"],
+            "fwd_byts_b_avg": flow_bytes.get_bytes_per_bulk(PacketDirection.FORWARD),
+            "fwd_pkts_b_avg": flow_bytes.get_packets_per_bulk(PacketDirection.FORWARD),
+            "bwd_byts_b_avg": flow_bytes.get_bytes_per_bulk(PacketDirection.REVERSE),
+            "bwd_pkts_b_avg": flow_bytes.get_packets_per_bulk(PacketDirection.REVERSE),
+            "fwd_blk_rate_avg": flow_bytes.get_bulk_rate(PacketDirection.FORWARD),
+            "bwd_blk_rate_avg": flow_bytes.get_bulk_rate(PacketDirection.REVERSE),
         }
 
         # Duplicated features
@@ -173,7 +195,7 @@ class Flow:
         """
         self.packets.append((packet, direction))
 
-        # self.update_flow_bulk(packet)
+        self.update_flow_bulk(packet, direction)
         self.update_subflow(packet)
 
         if self.start_timestamp != 0:
@@ -198,14 +220,25 @@ class Flow:
             self.protocol = packet.proto
 
     def update_subflow(self, packet):
+        """Update subflow
+
+        Args:
+            packet: Packet to be parse as subflow
+
+        """
         last_timestamp = (
             self.latest_timestamp if self.latest_timestamp != 0 else packet.time
         )
         if (packet.time - last_timestamp) > constants.CLUMP_TIMEOUT:
-            self.subflows += 1
             self.update_active_idle(packet.time - last_timestamp)
 
     def update_active_idle(self, current_time):
+        """Adds a packet to the current list of packets.
+
+        Args:
+            packet: Packet to be update active time
+
+        """
         if (current_time - self.last_active) > constants.ACTIVE_TIMEOUT:
             duration = self.last_active - self.start_active
             if duration > 0:
@@ -215,8 +248,82 @@ class Flow:
         else:
             self.last_active = current_time
 
-    def is_doh(self) -> bool:
-        return self.src_ip in constants.DOH_IPS or self.dest_ip in constants.DOH_IPS
+    def update_flow_bulk(self, packet, direction):
+        """Update bulk flow
+
+        Args:
+            packet: Packet to be parse as bulk
+
+        """
+        payload_size = len(PacketCount.get_payload(packet))
+        if payload_size == 0:
+            return
+        if direction == PacketDirection.FORWARD:
+            if self.backward_bulk_last_timestamp > self.forward_bulk_start_tmp:
+                self.forward_bulk_start_tmp = 0
+            if self.forward_bulk_start_tmp == 0:
+                self.forward_bulk_start_tmp = packet.time
+                self.forward_bulk_last_timestamp = packet.time
+                self.forward_bulk_count_tmp = 1
+                self.forward_bulk_size_tmp = payload_size
+            else:
+                if (
+                    packet.time - self.forward_bulk_last_timestamp
+                ) > constants.CLUMP_TIMEOUT:
+                    self.forward_bulk_start_tmp = packet.time
+                    self.forward_bulk_last_timestamp = packet.time
+                    self.forward_bulk_count_tmp = 1
+                    self.forward_bulk_size_tmp = payload_size
+                else:  # Add to bulk
+                    self.forward_bulk_count_tmp += 1
+                    self.forward_bulk_size_tmp += payload_size
+                    if self.forward_bulk_count_tmp == constants.BULK_BOUND:
+                        self.forward_bulk_count += 1
+                        self.forward_bulk_packet_count += self.forward_bulk_count_tmp
+                        self.forward_bulk_size += self.forward_bulk_size_tmp
+                        self.forward_bulk_duration += (
+                            packet.time - self.forward_bulk_start_tmp
+                        )
+                    elif self.forward_bulk_count_tmp > constants.BULK_BOUND:
+                        self.forward_bulk_packet_count += 1
+                        self.forward_bulk_size += payload_size
+                        self.forward_bulk_duration += (
+                            packet.time - self.forward_bulk_last_timestamp
+                        )
+                    self.forward_bulk_last_timestamp = packet.time
+        else:
+            if self.forward_bulk_last_timestamp > self.backward_bulk_start_tmp:
+                self.backward_bulk_start_tmp = 0
+            if self.backward_bulk_start_tmp == 0:
+                self.backward_bulk_start_tmp = packet.time
+                self.backward_bulk_last_timestamp = packet.time
+                self.backward_bulk_count_tmp = 1
+                self.backward_bulk_size_tmp = payload_size
+            else:
+                if (
+                    packet.time - self.backward_bulk_last_timestamp
+                ) > constants.CLUMP_TIMEOUT:
+                    self.backward_bulk_start_tmp = packet.time
+                    self.backward_bulk_last_timestamp = packet.time
+                    self.backward_bulk_count_tmp = 1
+                    self.backward_bulk_size_tmp = payload_size
+                else:  # Add to bulk
+                    self.backward_bulk_count_tmp += 1
+                    self.backward_bulk_size_tmp += payload_size
+                    if self.backward_bulk_count_tmp == constants.BULK_BOUND:
+                        self.backward_bulk_count += 1
+                        self.backward_bulk_packet_count += self.backward_bulk_count_tmp
+                        self.backward_bulk_size += self.backward_bulk_size_tmp
+                        self.backward_bulk_duration += (
+                            packet.time - self.backward_bulk_start_tmp
+                        )
+                    elif self.backward_bulk_count_tmp > constants.BULK_BOUND:
+                        self.backward_bulk_packet_count += 1
+                        self.backward_bulk_size += payload_size
+                        self.backward_bulk_duration += (
+                            packet.time - self.backward_bulk_last_timestamp
+                        )
+                    self.backward_bulk_last_timestamp = packet.time
 
     @property
     def duration(self):

tests/test_features.py

@@ -1,4 +1,4 @@
-from scapy.all import IP, TCP, UDP, ICMP, Ether
+from scapy.all import IP, TCP, UDP, ICMP
 from cicflowmeter.features.context.packet_flow_key import get_packet_flow_key
 from cicflowmeter.features.context.packet_direction import PacketDirection
 from cicflowmeter.flow import Flow
@@ -33,9 +33,90 @@ def mock_flow_data(mock_flow):
     return data
 
 
-def test_flow_data(mock_flow_data):
-    print(mock_flow_data)
-    assert len(mock_flow_data) > 70
+def test_features(mock_flow_data):
+    expected_keys = [
+        "dst_port",
+        "protocol",
+        "timestamp",
+        "flow_duration",
+        "tot_fwd_pkts",
+        "tot_bwd_pkts",
+        "totlen_fwd_pkts",
+        "totlen_bwd_pkts",
+        "fwd_pkt_len_max",
+        "fwd_pkt_len_min",
+        "fwd_pkt_len_mean",
+        "fwd_pkt_len_std",
+        "bwd_pkt_len_max",
+        "bwd_pkt_len_min",
+        "bwd_pkt_len_mean",
+        "bwd_pkt_len_std",
+        "flow_byts_s",
+        "flow_pkts_s",
+        "flow_iat_mean",
+        "flow_iat_std",
+        "flow_iat_max",
+        "flow_iat_min",
+        "fwd_iat_tot",
+        "fwd_iat_mean",
+        "fwd_iat_std",
+        "fwd_iat_max",
+        "fwd_iat_min",
+        "bwd_iat_tot",
+        "bwd_iat_mean",
+        "bwd_iat_std",
+        "bwd_iat_max",
+        "bwd_iat_min",
+        "fwd_psh_flags",
+        "bwd_psh_flags",
+        "fwd_urg_flags",
+        "bwd_urg_flags",
+        "fwd_header_len",
+        "bwd_header_len",
+        "fwd_pkts_s",
+        "bwd_pkts_s",
+        "pkt_len_min",
+        "pkt_len_max",
+        "pkt_len_mean",
+        "pkt_len_std",
+        "pkt_len_var",
+        "fin_flag_cnt",
+        "syn_flag_cnt",
+        "rst_flag_cnt",
+        "psh_flag_cnt",
+        "ack_flag_cnt",
+        "urg_flag_cnt",
+        "cwe_flag_count",
+        "ece_flag_cnt",
+        "down_up_ratio",
+        "pkt_size_avg",
+        "fwd_seg_size_avg",
+        "bwd_seg_size_avg",
+        "fwd_byts_b_avg",
+        "fwd_pkts_b_avg",
+        "fwd_blk_rate_avg",
+        "bwd_byts_b_avg",
+        "bwd_pkts_b_avg",
+        "bwd_blk_rate_avg",
+        "subflow_fwd_pkts",
+        "subflow_fwd_byts",
+        "subflow_bwd_pkts",
+        "subflow_bwd_byts",
+        "init_fwd_win_byts",
+        "init_bwd_win_byts",
+        "fwd_act_data_pkts",
+        "fwd_seg_size_min",
+        "active_mean",
+        "active_std",
+        "active_max",
+        "active_min",
+        "idle_mean",
+        "idle_std",
+        "idle_max",
+        "idle_min",
+    ]
+    for expected in expected_keys:
+        assert expected in mock_flow_data.keys()
 
 
 def test_packet_flow_key(mock_packet):

tox.ini

@@ -4,7 +4,7 @@
 # and then run "tox" from this directory.
 
 [tox]
-envlist = py37, linter, pypi
+envlist = py37, linter
 
 [testenv]
 deps =