Better handling of how slips stops and improve in memory storage of flows (Fix slips getting OOM killed)

.secrets.baseline

@@ -149,7 +149,7 @@
         "filename": "config/slips.yaml",
         "hashed_secret": "4cac50cee3ad8e462728e711eac3e670753d5016",
         "is_verified": false,
-        "line_number": 223
+        "line_number": 268
       }
     ],
     "dataset/test14-malicious-zeek-dir/http.log": [
@@ -7185,5 +7185,5 @@
       }
     ]
   },
-  "generated_at": "2026-01-29T18:21:33Z"
+  "generated_at": "2026-03-02T22:46:58Z"
 }

managers/process_manager.py

@@ -22,8 +22,6 @@
 from typing import (
     List,
     Tuple,
-    Dict,
-    Iterable,
 )
 
 from exclusiveprocess import (
@@ -58,18 +56,19 @@ def __init__(self, main):
         # Can be used by signal handlers before startup finishes.
         self.processes: List[Process] = []
 
-        # this is the queue that will be used by the input proces
+        # this is the queue that will be used by the input process
         # to pass flows to the profiler
-        # this max size is decided based on the avg size of each flow and
-        # tha max memory (4g) that this queue is allowed to use
-        self.profiler_queue = Queue(maxsize=50000)
-        self.termination_event: Event = Event()
+        # this max size is decided based on the avg size of each flow (650
+        # bytes), and the max memory that this queue is allowed to
+        # use (1GB), so 1321528 bytes will be 2033 flows in queue at max
+        self.profiler_queue = Queue(maxsize=1321528)
+        self.termination_event = Event()
         # to make sure we only warn the user once about
         # the pending modules
         self.warning_printed_once = False
         # this one has its own termination event because we want it to
         # shutdown at the very end of all other slips modules.
-        self.evidence_handler_termination_event: Event = Event()
+        self.evidence_handler_termination_event = Event()
         self.stopped_modules = []
         # used to stop slips when these 2 are done
         # since the semaphore count is zero, slips.py will wait until another
@@ -85,6 +84,9 @@ def __init__(self, main):
         # and inout stops and renders the profiler queue useless and profiler
         # cant get more lines anymore!
         self.is_profiler_done_event = Event()
+        # is set by the input process to indicate no more flows are coming
+        # so profiler can safely begin shutdown/joins.
+        self.is_input_done_event = Event()
         self.read_config()
 
     def read_config(self):
@@ -122,6 +124,7 @@ def start_profiler_process(self):
             is_profiler_done=self.is_profiler_done,
             profiler_queue=self.profiler_queue,
             is_profiler_done_event=self.is_profiler_done_event,
+            is_input_done_event=self.is_input_done_event,
         )
         profiler_process.start()
         self.main.print(
@@ -173,6 +176,7 @@ def start_input_process(self):
             zeek_dir=self.main.zeek_dir,
             line_type=self.main.line_type,
             is_profiler_done_event=self.is_profiler_done_event,
+            is_input_done_event=self.is_input_done_event,
         )
         input_process.start()
         self.main.print(
@@ -570,7 +574,8 @@ def get_hitlist_in_order(self) -> Tuple[List[Process], List[Process]]:
                 self.main.db.get_pid_of("Exporting Alerts")
             )
         # remove all None PIDs. this happens when a module in that list
-        # isnt started in the current run.
+        # isnt started in the current run. e.g. virustotal module starts then
+        # stops immediately if no API is found. so its pid will be None.
         pids_to_kill_last: List[int] = [
             pid for pid in pids_to_kill_last if pid is not None
         ]

managers/redis_manager.py

@@ -39,6 +39,13 @@ def __init__(self, main):
         self.end_port = 32850
         self.running_logfile = "running_slips_info.txt"
 
+    def _clear_cached_redis_instance(self, port: int) -> None:
+        """
+        Ensure RedisDB singleton cache doesn't block later startup flushes.
+        """
+        if port in RedisDB.instances:
+            del RedisDB.instances[port]
+
     def get_start_port(self):
         return self.start_port
 
@@ -386,6 +393,9 @@ def get_redis_port(self) -> int:
                         ):
                             self.main.terminate_slips()
                             return
+                    # allow the DBManager obj created in main.py to reconnect
+                    # and flush if needed
+                    self._clear_cached_redis_instance(redis_port)
 
         elif self.main.args.multiinstance:
             redis_port = self.get_random_redis_port()
@@ -413,6 +423,8 @@ def get_redis_port(self) -> int:
                         f"{redis_port}."
                     )
                     self.main.terminate_slips()
+            # allow main DBManager to reconnect and flush if needed
+            self._clear_cached_redis_instance(redis_port)
 
         return redis_port
 

modules/anomaly_detection_https/anomaly_detection_https.py

@@ -117,7 +117,9 @@ def update_min_std_floor(self):
         candidate = self.floor_scale * max(q10, sigma_mad, self.floor_min)
         candidate = min(self.floor_max, max(self.floor_min, candidate))
         beta = min(1.0, max(0.0, self.floor_update_beta))
-        self.min_std_floor = (1.0 - beta) * self.min_std_floor + beta * candidate
+        self.min_std_floor = (
+            1.0 - beta
+        ) * self.min_std_floor + beta * candidate
 
     def zscore(self, value: float) -> float:
         std = math.sqrt(max(self.var, self.min_std_floor * self.min_std_floor))
@@ -162,8 +164,6 @@ class AnomalyDetectionHTTPS(IModule):
     authors = ["Sebastian Garcia"]
 
     def init(self):
-        self.c1 = self.db.subscribe("new_ssl")
-        self.channels = {"new_ssl": self.c1}
         self.classifier = FlowClassifier()
         self.read_configuration()
         self.operational_log_path = os.path.join(
@@ -205,6 +205,10 @@ def init(self):
                 "ADWIN requested but river is not installed; using pre-ADWIN drift logic.",
             )
 
+    def subscribe_to_channels(self):
+        self.c1 = self.db.subscribe("new_ssl")
+        self.channels = {"new_ssl": self.c1}
+
     def read_configuration(self):
         conf = ConfigParser()
         self.training_hours = conf.https_anomaly_training_hours()
@@ -221,9 +225,7 @@ def read_configuration(self):
         self.ja3_min_variants_per_server = (
             conf.https_anomaly_ja3_min_variants_per_server()
         )
-        self.requested_use_adwin_drift = (
-            conf.https_anomaly_use_adwin_drift()
-        )
+        self.requested_use_adwin_drift = conf.https_anomaly_use_adwin_drift()
         self.adwin_delta = conf.https_anomaly_adwin_delta()
         self.adwin_clock = conf.https_anomaly_adwin_clock()
         self.adwin_grace_period = conf.https_anomaly_adwin_grace_period()
@@ -318,9 +320,7 @@ def log_event(
         reset = "\033[0m" if self.log_colors else ""
         wall_clock = self._ts_to_iso()
         traffic_clock = (
-            self._ts_to_iso(traffic_ts)
-            if traffic_ts is not None
-            else "n/a"
+            self._ts_to_iso(traffic_ts) if traffic_ts is not None else "n/a"
         )
         metrics_json = json.dumps(metrics, sort_keys=True)
         line = (
@@ -329,7 +329,9 @@ def log_event(
         )
         if color:
             line = f"{color}{line}{reset}"
-        with open(self.operational_log_path, "a", encoding="utf-8") as log_file:
+        with open(
+            self.operational_log_path, "a", encoding="utf-8"
+        ) as log_file:
             log_file.write(f"{line}\n")
 
     @staticmethod
@@ -344,7 +346,9 @@ def to_float(value, default=0.0) -> float:
         except (TypeError, ValueError):
             return float(default)
 
-    def get_traffic_ts(self, flow, fallback_ts: Optional[float] = None) -> float:
+    def get_traffic_ts(
+        self, flow, fallback_ts: Optional[float] = None
+    ) -> float:
         """
         Returns traffic timestamp from flow.starttime.
         Detection windows must use traffic time, not host wall-clock time.
@@ -466,7 +470,9 @@ def evidence_ts_from_traffic_ts(ts: float) -> str:
         )
 
     @staticmethod
-    def threat_level_from_confidence_level(confidence_level: str) -> ThreatLevel:
+    def threat_level_from_confidence_level(
+        confidence_level: str,
+    ) -> ThreatLevel:
         # Requested policy:
         # - confidence low/medium -> threat level low
         # - confidence high -> threat level medium
@@ -500,7 +506,7 @@ def build_victim(
             )
         return None
 
-    def emit_anomaly_evidence(
+    def set_anomaly_evidence(
         self,
         profileid: str,
         twid_number: int,
@@ -571,7 +577,11 @@ def emit_anomaly_evidence(
                 f"reason={reason_name}; value={value}; why={why}"
             )
 
-        reasons_text = " | ".join(reason_parts) if reason_parts else "reason=Unknown; value=; why=not provided"
+        reasons_text = (
+            " | ".join(reason_parts)
+            if reason_parts
+            else "reason=Unknown; value=; why=not provided"
+        )
         description = (
             f"HTTPS anomaly: type={kind}; confidence={confidence.get('level')} "
             f"({confidence_score:.3f}); {reasons_text}."
@@ -699,9 +709,8 @@ def finalize_hour_bucket(self, profileid: str, state: HostState):
         ssl_flows = float(bucket.ssl_flows)
         known_server_avg_bytes = 0.0
         if bucket.known_servers_flow_count > 0:
-            known_server_avg_bytes = (
-                bucket.known_servers_total_bytes
-                / float(bucket.known_servers_flow_count)
+            known_server_avg_bytes = bucket.known_servers_total_bytes / float(
+                bucket.known_servers_flow_count
             )
 
         features = {
@@ -743,7 +752,9 @@ def finalize_hour_bucket(self, profileid: str, state: HostState):
             for feature_name, value in features.items():
                 z = z_by_feature.get(feature_name, 0.0)
                 if z >= self.hourly_zscore_threshold:
-                    model = self.get_or_create_hourly_model(state, feature_name)
+                    model = self.get_or_create_hourly_model(
+                        state, feature_name
+                    )
                     hourly_anomalies.append(
                         {
                             "feature": feature_name,
@@ -787,7 +798,7 @@ def finalize_hour_bucket(self, profileid: str, state: HostState):
                         "anomalies": hourly_anomalies,
                     },
                 )
-                self.emit_anomaly_evidence(
+                self.set_anomaly_evidence(
                     profileid=profileid,
                     twid_number=state.last_twid,
                     traffic_ts=bucket.start_ts,
@@ -996,7 +1007,8 @@ def process_ssl_event(
         twid_number: int,
     ):
         ts = self.get_traffic_ts(
-            ssl_flow, fallback_ts=self.to_float(conn_info.get("starttime"), 0.0)
+            ssl_flow,
+            fallback_ts=self.to_float(conn_info.get("starttime"), 0.0),
         )
         state = self.ensure_hour_bucket(profileid, ts)
         state.last_twid = twid_number
@@ -1111,7 +1123,7 @@ def process_ssl_event(
                     "flow_anomalies": flow_anomalies,
                 },
             )
-            self.emit_anomaly_evidence(
+            self.set_anomaly_evidence(
                 profileid=profileid,
                 twid_number=twid_number,
                 traffic_ts=ts,
@@ -1244,9 +1256,7 @@ def process_ssl_event(
                     "flow_raw_signals": flow_raw_signals,
                     "alpha": alpha,
                     "fit_method": (
-                        "welford_online_moments"
-                        if alpha is None
-                        else "ewma"
+                        "welford_online_moments" if alpha is None else "ewma"
                     ),
                 },
             )

modules/arp/arp.py

@@ -34,12 +34,6 @@ class ARP(IModule):
     authors = ["Alya Gomaa"]
 
     def init(self):
-        self.c1 = self.db.subscribe("new_arp")
-        self.c2 = self.db.subscribe("tw_closed")
-        self.channels = {
-            "new_arp": self.c1,
-            "tw_closed": self.c2,
-        }
         self.read_configuration()
         self.classifier = FlowClassifier()
         # this dict will categorize arp requests by profileid_twid
@@ -68,6 +62,14 @@ def init(self):
         self.is_zeek_running: bool = self.is_running_zeek()
         self.evidence_filter = ARPEvidenceFilter(self.conf, self.args, self.db)
 
+    def subscribe_to_channels(self):
+        self.c1 = self.db.subscribe("new_arp")
+        self.c2 = self.db.subscribe("tw_closed")
+        self.channels = {
+            "new_arp": self.c1,
+            "tw_closed": self.c2,
+        }
+
     def read_configuration(self):
         conf = ConfigParser()
         self.home_network = conf.home_network_ranges

modules/arp_poisoner/arp_poisoner.py

@@ -31,12 +31,6 @@ class ARPPoisoner(IModule):
     authors = ["Alya Gomaa"]
 
     def init(self):
-        self.c1 = self.db.subscribe("new_blocking")
-        self.c2 = self.db.subscribe("tw_closed")
-        self.channels = {
-            "new_blocking": self.c1,
-            "tw_closed": self.c2,
-        }
         self._time_since_last_repoison = {}
         self._time_since_last_internet_cut = {}
         self.log_file_path = os.path.join(self.output_dir, "arp_poisoning.log")
@@ -60,6 +54,14 @@ def init(self):
         # keeps track of which interface were blocked ips attacking on
         self.ip_interface_map = {}
 
+    def subscribe_to_channels(self):
+        self.c1 = self.db.subscribe("new_blocking")
+        self.c2 = self.db.subscribe("tw_closed")
+        self.channels = {
+            "new_blocking": self.c1,
+            "tw_closed": self.c2,
+        }
+
     def log(self, text):
         """Logs the given text to the blocking log file"""
         with self.blocking_logfile_lock:

modules/blocking/blocking.py

@@ -29,12 +29,6 @@ class Blocking(IModule):
     authors = ["Sebastian Garcia, Alya Gomaa"]
 
     def init(self):
-        self.c1 = self.db.subscribe("new_blocking")
-        self.c2 = self.db.subscribe("tw_closed")
-        self.channels = {
-            "new_blocking": self.c1,
-            "tw_closed": self.c2,
-        }
         if platform.system() == "Darwin":
             self.print("Mac OS blocking is not supported yet.")
             sys.exit()
@@ -54,6 +48,14 @@ def init(self):
         self.ap_info: None | Dict[str, str] = self.db.get_ap_info()
         self.is_running_in_ap_mode = True if self.ap_info else False
 
+    def subscribe_to_channels(self):
+        self.c1 = self.db.subscribe("new_blocking")
+        self.c2 = self.db.subscribe("tw_closed")
+        self.channels = {
+            "new_blocking": self.c1,
+            "tw_closed": self.c2,
+        }
+
     def log(self, text: str):
         """Logs the given text to the blocking log file"""
         with self.blocking_logfile_lock:

modules/cesnet/cesnet.py

@@ -29,11 +29,13 @@ class CESNET(IModule):
 
     def init(self):
         self.read_configuration()
+        self.stop_module = False
+
+    def subscribe_to_channels(self):
         self.c1 = self.db.subscribe("export_evidence")
         self.channels = {
             "export_evidence": self.c1,
         }
-        self.stop_module = False
 
     def read_configuration(self):
         """Read importing/exporting preferences from slips.yaml"""