Fix setting data exfiltration and SSH successful evidence

modules/flowalerts/flowalerts.py

@@ -1,6 +1,4 @@
 import contextlib
-from typing import List
-
 from slips_files.common.imports import *
 
 import json
@@ -389,55 +387,66 @@ def check_pastebin_download(
             # maybe an empty file is downloaded
             return False
 
+    def get_sent_bytes(self, all_flows: Dict[str, dict]) \
+            -> Dict[str, Tuple[int, List[str], str]] :
+        """
+        Returns a dict of sent bytes to all ips in the all_flows dict
+         {
+            contacted_ip: (
+                sum_of_mbs_sent,
+                [uids],
+                last_ts_of_flow_containging_this_contacted_ip
+            )
+        }
+         """
+        bytes_sent = {}
+        for uid, flow in all_flows.items():
+            daddr = flow['daddr']
+            sbytes: int = flow.get('sbytes', 0)
+            ts: str = flow.get('starttime', '')
+
+            if self.is_ignored_ip_data_upload(daddr) or not sbytes:
+                continue
+
+            if daddr in bytes_sent:
+                mbs_sent, uids, _= bytes_sent[daddr]
+                mbs_sent += sbytes
+                uids.append(uid)
+                bytes_sent[daddr] = (mbs_sent, uids, ts)
+            else:
+                bytes_sent[daddr] = (sbytes, [uid], ts)
 
+        return bytes_sent
+
     def detect_data_upload_in_twid(self, profileid, twid):
         """
         For each contacted ip in this twid,
         check if the total bytes sent to this ip is >= data_exfiltration_threshold
         """
-        def get_sent_bytes(all_flows: dict):
-            """Returns a dict of sent bytes to all ips {contacted_ip: (mbs_sent, [uids])}"""
-            bytes_sent = {}
-            for uid, flow in all_flows.items():
-                daddr = flow['daddr']
-                sbytes: int = flow.get('sbytes', 0)
-
-                if self.is_ignored_ip_data_upload(daddr) or not sbytes:
-                    continue
-
-                if daddr in bytes_sent:
-                    mbs_sent, uids = bytes_sent[daddr]
-                    mbs_sent += sbytes
-                    uids.append(uid)
-                    bytes_sent[daddr] = (mbs_sent, uids)
-                else:
-                    bytes_sent[daddr] = (sbytes, [uid])
-
-            return bytes_sent
-
         all_flows: Dict[str, dict] = self.db.get_all_flows_in_profileid(
             profileid
         )
         if not all_flows:
             return
 
-        bytes_sent: dict = get_sent_bytes(all_flows)
+        bytes_sent: Dict[str, Tuple[int, List[str], str]]
+        bytes_sent = self.get_sent_bytes(all_flows)
 
         for ip, ip_info in bytes_sent.items():
-            ip_info: Tuple[int, List[str]]
-            uids = ip_info[1]
-            bytes_uploaded = ip_info[0]
+            ip_info: Tuple[int, List[str], str]
+            bytes_uploaded, uids, ts = ip_info
 
             mbs_uploaded = utils.convert_to_mb(bytes_uploaded)
             if mbs_uploaded < self.data_exfiltration_threshold:
                 continue
-
+                
             self.set_evidence.data_exfiltration(
                 ip,
                 mbs_uploaded,
                 profileid,
                 twid,
                 uids,
+                ts
             )
 
 
@@ -738,7 +747,6 @@ def check_dns_without_connection(
     ):
         """
         Makes sure all cached DNS answers are used in contacted_ips
-        :param contacted_ips:  dict of ips used in a specific tw {ip: uid}
         """
         ## - All reverse dns resolutions
         ## - All .local domains
@@ -863,7 +871,7 @@ def detect_successful_ssh_by_zeek(self, uid, timestamp, profileid, twid):
             )
             daddr = ssh_flow_dict['daddr']
             saddr = ssh_flow_dict['saddr']
-            size = ssh_flow_dict['allbytes']
+            size = ssh_flow_dict['sbytes'] + ssh_flow_dict['dbytes']
             self.set_evidence.ssh_successful(
                 twid,
                 saddr,
@@ -976,6 +984,7 @@ def detect_incompatible_CN(
         """
         if not issuer:
             return False
+
         found_org_in_cn = ''
         for org in utils.supported_orgs:
             if org not in issuer.lower():
@@ -1115,7 +1124,7 @@ def check_invalid_dns_answers(
             if answer in invalid_answers and domain != "localhost":
                 # blocked answer found
                 self.set_evidence.invalid_dns_answer(
-                    domain, answer, daddr, profileid, twid, stime, uid
+                    domain, answer, profileid, twid, stime, uid
                 )
                 # delete answer from redis cache to prevent
                 # associating this dns answer with this domain/query and
@@ -1663,8 +1672,8 @@ def check_non_http_port_80_conns(
     def check_GRE_tunnel(self, tunnel_info: dict):
         """
         Detects GRE tunnels
-        @param tunnel_flow: dict containing tunnel zeek flow
-        @return: None
+        :param tunnel_info: dict containing tunnel zeek flow
+        :return: None
         """
         tunnel_flow = tunnel_info['flow']
         tunnel_type = tunnel_flow['tunnel_type']
@@ -2107,7 +2116,7 @@ def main(self):
 
         if msg := self.get_msg('tw_closed'):
             profileid_tw = msg['data'].split('_')
-            profileid = f'{profileid_tw[0]}_{profileid_tw[1]}',
+            profileid = f'{profileid_tw[0]}_{profileid_tw[1]}'
             twid = profileid_tw[-1]
             self.detect_data_upload_in_twid(profileid, twid)
 

modules/flowalerts/set_evidence.py

@@ -697,7 +697,7 @@ def conn_to_private_ip(
             self,
             proto: str,
             daddr: str,
-            dport: str,
+            dport: int,
             saddr: str,
             twid: str,
             uid: str,
@@ -838,7 +838,7 @@ def ssh_successful(
             twid: str,
             saddr: str,
             daddr: str,
-            size,
+            size: int,
             uid: str,
             timestamp: str,
             by='',
@@ -848,6 +848,7 @@ def ssh_successful(
         This is not strictly a detection, but we don't have
         a better way to show it.
         The threat_level is 0.01 to show that this is not a detection
+        :param size: src and dst bytes sent and recieved
         """
 
         confidence: float = 0.8
@@ -857,7 +858,7 @@ def ssh_successful(
         ip_identification: str = self.db.get_ip_identification(daddr)
         description: str = (
             f'SSH successful to IP {daddr}. {ip_identification}. '
-            f'From IP {saddr}. Size: {str(size)}. Detection model {by}.'
+            f'From IP {saddr}. Sent bytes: {str(size)}. Detection model {by}.'
             f' Confidence {confidence}'
         )
 

slips_files/core/database/redis_db/profile_handler.py

@@ -309,10 +309,6 @@ def add_port(
         ip = str(flow.daddr)
         spkts = flow.spkts
         state_hist = flow.state_hist if hasattr(flow, "state_hist") else ""
-        # dpkts = columns['dpkts']
-        # daddr = columns['daddr']
-        # saddr = columns['saddr']
-        # sbytes = columns['sbytes']
 
         if "^" in state_hist:
             # The majority of the FP with horizontal port scan detection happen because a