when a tw is closed, delete all info from the db starting from this tw-2, (always keep track of 2 tws only)

Author: AlyaGomaa

slips_files/core/database/redis_db/profile_handler.py

@@ -1064,17 +1064,59 @@ def check_tw_to_close(self, close_all=False):
         if not profiles_tws_to_close:
             return
 
-        # Mark the TWs as closed so module can work on its data
-        pipeline = self.r.pipeline()
+        # Mark the TWs as closed so modules can work on its data
+        pipe = self.r.pipeline()
         for profile_tw_to_close in profiles_tws_to_close:
-            pipeline.sadd("ClosedTW", profile_tw_to_close)
-            pipeline.zrem(
-                self.constants.MODIFIED_TIMEWINDOWS, profile_tw_to_close
+            pipe.zrem(self.constants.MODIFIED_TIMEWINDOWS, profile_tw_to_close)
+            pipe = self.publish(
+                "tw_closed", profile_tw_to_close, pipeline=pipe
             )
-            pipeline = self.publish(
-                "tw_closed", profile_tw_to_close, pipeline=pipeline
+            pipe = self._delete_past_timewindows(profile_tw_to_close, pipe)
+        pipe.execute()
+
+    def _delete_past_timewindows(self, closed_profile_tw: str, pipe):
+        """
+        Deletes the past timewindows data from redis, starting from the
+        given tw-1, so that redis only has info about the current
+        timewindow and the one before it
+
+        why do we keep 2 tws instead of the current one in redis? see PR
+        #1765 in slips repo
+
+        :param closed_profile_tw: a str like profile_8.8.8.8_timewindow7
+        """
+        # todo handle flows that have a ts in the very future that cause
+        #  this func to del all tws!!
+        try:
+            profile, ip, tw = closed_profile_tw.split("_")
+            closed_tw = int(tw.replace("timewindow", ""))
+        except ValueError:
+            self.print(
+                f"Unable to delete old Timewindwos info from"
+                f" {closed_profile_tw}"
             )
-        pipeline.execute()
+            return pipe
+
+        if closed_tw < 2:
+            # slips needs to always remember 2 tws, now tws to delete now
+            return pipe
+
+        profileid = f"{profile}_{ip}"
+        # to avoid deleting so many keys at once which causes mem spikes
+        BATCH = 500
+        for tw_to_close in range(closed_tw - 2, -1, -1):
+            for i, key in enumerate(
+                self.r.scan_iter(
+                    match=f"{profileid}_timewindow{tw_to_close}", count=1000
+                )
+            ):
+                pipe.unlink(key)
+
+                if i % BATCH == 0:
+                    pipe.execute()
+                    pipe = self.r.pipeline()
+
+        return pipe
 
     def mark_profile_tw_as_modified(
         self, profileid, twid, timestamp, pipe: Pipeline = None

slips_files/core/database/redis_db/scan_detections_db.py

@@ -125,12 +125,10 @@ def _update_portscan_index_hash(
         # if no first seen ts is set, then this flow is the first seen
         key = f"{base}:first_seen"
         pipe.zadd(key, {ip: flow.starttime}, nx=True)
-        pipe.expire(key, self.tw_width, nx=True)
 
         key = f"{base}:last_seen"
         # last seen is now. this flow.
         pipe.zadd(key, {ip: flow.starttime})
-        pipe.expire(key, self.tw_width, nx=True)
 
         return pipe
 
@@ -397,7 +395,6 @@ def _store_vertical_portscan_info(
             f":{str_proto}:not_estab:{target_ip}:dstports"
         )
         pipe.hincrby(key, flow.dport, int(flow.pkts))
-        pipe.expire(key, self.tw_width, nx=True)
         # increment the total pkts sent to this target ip on this
         # proto so slips can retreieve it in O(1) when setting and
         # evidence
@@ -407,7 +404,6 @@ def _store_vertical_portscan_info(
             f"{target_ip}:dstports:tot_pkts_sum"
         )
         pipe.incrby(key, int(flow.spkts))
-        pipe.expire(key, self.tw_width, nx=True)
 
         # we keep an index hash of target_ips to be able to access the
         # diff variants of the key above using them
@@ -430,7 +426,6 @@ def _store_horizontal_portscan_info(
                 f"{str_proto}:not_estab:dstports:total_packets"
             )
             pipe.hincrby(key, flow.dport, int(flow.pkts))
-            pipe.expire(key, self.tw_width, nx=True)
 
             # ZSET
             # profile_tw:[tcp|udp]:not_estab:dport:
@@ -445,7 +440,6 @@ def _store_horizontal_portscan_info(
             # To make sure the stored ts is the first seen ts of this
             # daddr, we use nx=True, so if a daddr is present we dont zadd
             pipe.zadd(key, {flow.daddr: flow.starttime}, nx=True)
-            pipe.expire(key, self.tw_width, nx=True)
 
         return pipe
 
@@ -458,21 +452,17 @@ def _store_conn_to_multiple_ports_info(
         if role == role.CLIENT:
             key = f"{profileid}_{twid}:tcp:est:dstips"
             pipe.zadd(key, {flow.daddr: flow.starttime}, nx=True)
-            pipe.expire(key, self.tw_width, nx=True)
 
             key = f"{profileid}_{twid}:tcp:est:{flow.daddr}:dstports"
             pipe.hset(key, flow.dport, flow.uid)
-            pipe.expire(key, self.tw_width, nx=True)
 
         elif role == role.SERVER:
             client_profileid = ProfileID(ip=flow.saddr)
             key = f"{client_profileid}_{twid}:tcp:est:dstips"
             pipe.zadd(key, {flow.saddr: flow.starttime}, nx=True)
-            pipe.expire(key, self.tw_width, nx=True)
 
             key = f"{client_profileid}_{twid}:tcp:est:{flow.saddr}:dstports"
             pipe.hset(key, flow.dport, flow.uid)
-            pipe.expire(key, self.tw_width, nx=True)
         return pipe
 
     def _store_flow_info_if_needed_by_detection_modules(