Merge pull request #1438 from stratosphereips/alya/fix_slips_not_shutting_down_gracefully

Fix slips not shutting down gracefully

Author: AlyaGomaa

docs/detection_modules.md

@@ -255,9 +255,16 @@ You can add your own YARA rule in ```modules/leak_detector/yara_rules/rules``` a
 
 ## Blocking Module
 
-To enable blocking in slips, start slips with the ```-p``` flag.
+Blocking in Slips is done for any IP that results in an alert. If an IP is detected as malicious and is blocked,
+it stays blocked forever, unless it is unblocked manually.
 
-This feature is only supported in linux using iptables.
+The feature of unblocking IPs after a while is not supported yet.
+
+The blocking is done using iptables, and the blocked IPs are stored in the database for future reference.
+
+Blocking is disabled by default. To enable blocking in slips, start slips with the ```-p``` flag.
+
+This feature is only supported in linux using iptables when running on an interface.
 
 ## Exporting Alerts Module
 
@@ -850,7 +857,7 @@ abuse.ch -> Used by urlhaus for getting info about contacted domains and downloa
 If you want to contribute: improve existing Slips detection modules or implement your own detection modules, see section :doc:`Contributing <contributing>`.
 
 
-## Zeek Scripts
+# Zeek Scripts
 
 Slips is shipped with it's own custom zeek scripts to be able to extend zeek functionality and
 customize the detections

docs/exporting.md

@@ -1,4 +1,4 @@
-# Exporting
+# Exporting Slips Alerts
 
 Slips supports exporting alerts to other systems using different modules (ExportingAlerts, CESNET sharing etc.)
 

docs/features.md

@@ -772,11 +772,11 @@ the update manager if they were changed on disk.
 
 ### IP Info Module
 
-The IP info module has several ways of getting information about an IP address, it includes:
+The IP info module has several ways of getting information about IP and MAC address, it includes:
 
 - ASN
 - Country by Geolocation
-- Given a MAC, its Vendor
+- MAC Vendors
 - Reverse DNS
 
 #### ASN

docs/flowalerts.md

@@ -1,4 +1,4 @@
-# Flow Alerts
+# Flow Alerts Module
 
 The module of flow alerts has several behavioral techniques to detect attacks by analyzing the content of each flow alone.
 

managers/process_manager.py

@@ -54,6 +54,9 @@ def __init__(self, main):
         # to pass flows to the profiler
         self.profiler_queue = Queue()
         self.termination_event: 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()
@@ -468,17 +471,18 @@ def get_hitlist_in_order(self) -> Tuple[List[Process], List[Process]]:
             pid for pid in pids_to_kill_last if pid is not None
         ]
 
+        # now get the process obj of each pid
         to_kill_first: List[Process] = []
         to_kill_last: List[Process] = []
         for process in self.processes:
             if process.pid in pids_to_kill_last:
                 to_kill_last.append(process)
-            else:
+            elif isinstance(process, multiprocessing.context.ForkProcess):
                 # skips the context manager of output.py, will close
                 # it manually later
                 # once all processes are closed
-                if isinstance(process, multiprocessing.context.ForkProcess):
-                    continue
+                continue
+            else:
                 to_kill_first.append(process)
 
         return to_kill_first, to_kill_last
@@ -677,12 +681,14 @@ def shutdown_gracefully(self):
                 print("\n" + "-" * 27)
             print("Stopping Slips")
 
-            # by default, 15 mins from this time, all modules should be killed
+            # by default, max 15 mins (taken from wait_for_modules_to_finish)
+            # from this time, all modules should be killed
             method_start_time = time.time()
 
             # how long to wait for modules to finish in minutes
             timeout: float = self.main.conf.wait_for_modules_to_finish()
-            timeout_seconds: float = timeout * 60
+            # convert to seconds
+            timeout *= 60
 
             # close all tws
             self.main.db.check_tw_to_close(close_all=True)
@@ -701,20 +707,15 @@ def shutdown_gracefully(self):
                     log_to_logfiles_only=True,
                 )
 
-                hitlist: Tuple[List[Process], List[Process]]
-                hitlist = self.get_hitlist_in_order()
-                to_kill_first: List[Process] = hitlist[0]
-                to_kill_last: List[Process] = hitlist[1]
+                to_kill_first: List[Process]
+                to_kill_last: List[Process]
+                to_kill_first, to_kill_last = self.get_hitlist_in_order()
 
                 self.termination_event.set()
 
-                # to make sure we only warn the user once about
-                # the pending modules
-                self.warning_printed_once = False
-
                 try:
                     # Wait timeout_seconds for all the processes to finish
-                    while time.time() - method_start_time < timeout_seconds:
+                    while time.time() - method_start_time < timeout:
                         (
                             to_kill_first,
                             to_kill_last,
@@ -733,7 +734,7 @@ def shutdown_gracefully(self):
                     reason = "User pressed ctr+c or Slips was killed by the OS"
                     graceful_shutdown = False
 
-                if time.time() - method_start_time >= timeout_seconds:
+                if time.time() - method_start_time >= timeout:
                     # getting here means we're killing them bc of the timeout
                     # not getting here means we're killing them bc of double
                     # ctr+c OR they terminated successfully

modules/arp/arp.py

@@ -55,7 +55,9 @@ def init(self):
             # thats one hour in seconds
             self.period_before_deleting = 3600
         self.timer_thread_arp_scan = threading.Thread(
-            target=self.wait_for_arp_scans, daemon=True
+            target=self.wait_for_arp_scans,
+            daemon=True,
+            name="timer_thread_arp_scan",
         )
         self.pending_arp_scan_evidence = Queue()
         self.alerted_once_arp_scan = False
@@ -88,7 +90,7 @@ def wait_for_arp_scans(self):
         """
         # this evidence is the one that triggered this thread
         scans_ctr = 0
-        while True:
+        while not self.should_stop():
             try:
                 evidence: dict = self.pending_arp_scan_evidence.get(
                     timeout=0.5
@@ -515,7 +517,7 @@ def clear_arp_logfile(self):
     def pre_main(self):
         """runs once before the main() is executed in a loop"""
         utils.drop_root_privs()
-        self.timer_thread_arp_scan.start()
+        utils.start_thread(self.timer_thread_arp_scan, self.db)
 
     def main(self):
         self.clear_arp_logfile()

modules/cesnet/cesnet.py

@@ -131,9 +131,11 @@ def export_evidence(self, evidence: dict):
         # and don't stop this module until the thread is done
         q = queue.Queue()
         self.sender_thread = threading.Thread(
-            target=self.wclient.sendEvents, args=[[evidence_in_idea], q]
+            target=self.wclient.sendEvents,
+            args=[[evidence_in_idea], q],
+            name="cesnet_sender_thread",
         )
-        self.sender_thread.start()
+        utils.start_thread(self.sender_thread, self.db)
         self.sender_thread.join()
         result = q.get()
 

modules/exporting_alerts/stix_exporter.py

@@ -28,14 +28,16 @@ def init(self):
             # To avoid duplicates in STIX_data.json
             self.added_ips = set()
             self.export_to_taxii_thread = threading.Thread(
-                target=self.schedule_sending_to_taxii_server, daemon=True
+                target=self.schedule_sending_to_taxii_server,
+                daemon=True,
+                name="stix_exporter_to_taxii_thread",
             )
 
     def start_exporting_thread(self):
         # This thread is responsible for waiting n seconds before
         # each push to the stix server
         # it starts the timer when the first alert happens
-        self.export_to_taxii_thread.start()
+        utils.start_thread(self.export_to_taxii_thread, self.db)
 
     @property
     def name(self):

modules/flowalerts/dns.py

@@ -5,6 +5,7 @@
 import json
 import math
 import queue
+import time
 from datetime import datetime
 from typing import (
     List,
@@ -13,7 +14,7 @@
 )
 import validators
 from multiprocessing import Queue
-from threading import Thread
+from threading import Thread, Event
 
 from slips_files.common.abstracts.flowalerts_analyzer import (
     IFlowalertsAnalyzer,
@@ -48,8 +49,9 @@ def init(self):
         self.dns_without_connection_timeout_checker_thread = Thread(
             target=self.check_dns_without_connection_timeout,
             daemon=True,
+            name="dns_without_connection_timeout_checker_thread",
         )
-
+        self.stop_event = Event()
         # used to pass the msgs this analyzer reciecved, to the
         # dns_without_connection_timeout_checker_thread.
         # the reason why we can just use .get_msg() there is because once
@@ -222,7 +224,6 @@ def is_any_flow_answer_contacted(self, profileid, twid, flow) -> bool:
         if flow.answers == ["-"]:
             # If no IPs are in the answer, we can not expect
             # the computer to connect to anything
-            # self.print(f'No ips in the answer, so ignoring')
             return True
 
         contacted_ips = self.db.get_all_contacted_ips_in_profileid_twid(
@@ -233,7 +234,6 @@ def is_any_flow_answer_contacted(self, profileid, twid, flow) -> bool:
         # one of these ips should be present in the contacted ips
         # check each one of the resolutions of this domain
         for ip in self.extract_ips_from_dns_answers(flow.answers):
-            # self.print(f'Checking if we have a connection to ip {ip}')
             if (
                 ip in contacted_ips
                 or self.is_connection_made_by_different_version(
@@ -342,10 +342,10 @@ def check_dns_without_connection_timeout(self):
         Does so by receiving every dns msg this analyzer receives. then we
         compare the ts of it to the ts of the flow we're waiting the 30
         mins for.
-        once we know the diff between them is >=30 mins we check for the
+        once we know that >=30 mins passed between them we check for the
         dns without connection evidence.
         The whole point is to give the connection 30 mins in zeek time to
-        arrive before alerting "dns wihtout conn".
+        arrive before setting "dns wihtout conn" evidence.
 
         - To avoid having thousands of flows in memory for 30 mins. we check
         every 10 mins for the connections, if not found we put it back to
@@ -354,16 +354,21 @@ def check_dns_without_connection_timeout(self):
         This function runs in its own thread
         """
         try:
-            while not self.flowalerts.should_stop():
-                if self.pending_dns_without_conn.empty():
-                    continue
+            while (
+                not self.flowalerts.should_stop()
+                and not self.stop_event.is_set()
+            ):
+                # if self.pending_dns_without_conn.empty():
+                #     time.sleep(1)
+                #     continue
 
                 # we just use it to know the zeek current ts to check if 30
                 # mins zeek time passed or not. we are not going to
                 # analyze it.
                 reference_flow = self.get_dns_flow_from_queue()
                 if not reference_flow:
                     # ok wait for more dns flows to be read by slips
+                    time.sleep(1)
                     continue
 
                 # back_to_queue will be used to store the flows we're
@@ -378,9 +383,12 @@ def check_dns_without_connection_timeout(self):
                 for flow in back_to_queue:
                     flow: Tuple[str, str, Any]
                     self.pending_dns_without_conn.put(flow)
+
         except KeyboardInterrupt:
             # the rest will be handled in shutdown_gracefully
             return
+        except Exception:
+            self.print_traceback()
 
     async def check_dns_without_connection(
         self, profileid, twid, flow, waited_for_the_conn=False
@@ -686,7 +694,16 @@ def check_different_localnet_usage(
 
     def shutdown_gracefully(self):
         self.check_dns_without_connection_of_all_pending_flows()
-        self.dns_without_connection_timeout_checker_thread.join()
+        self.stop_event.set()
+        self.dns_without_connection_timeout_checker_thread.join(30)
+        if self.dns_without_connection_timeout_checker_thread.is_alive():
+            self.flowalerts.print(
+                f"Problem shutting down "
+                f"dns_without_connection_timeout_checker_thread."
+                f"Flowalerts should_stop(): "
+                f"{self.flowalerts.should_stop()}"
+            )
+
         # close the queue
         # without this, queues are left in memory and flowalerts keeps
         # waiting for them forever
@@ -703,7 +720,9 @@ def pre_analyze(self):
         flowalerts' pre_main"""
         # we didnt put this in __init__ because it uses self.flowalerts
         # attributes that are not initialized yet in __init__
-        self.dns_without_connection_timeout_checker_thread.start()
+        utils.start_thread(
+            self.dns_without_connection_timeout_checker_thread, self.db
+        )
 
     async def analyze(self, msg):
         """

modules/p2ptrust/p2ptrust.py

@@ -117,7 +117,9 @@ def init(self, *args, **kwargs):
             self.print(f"Storing p2p.log in {self.pigeon_logfile}")
             # create the thread that rotates the p2p.log every 1d
             self.rotator_thread = threading.Thread(
-                target=self.rotate_p2p_logfile, daemon=True
+                target=self.rotate_p2p_logfile,
+                daemon=True,
+                name="p2p_rotator_thread",
             )
 
         self.storage_name = "IPsInfo"
@@ -648,7 +650,7 @@ def pre_main(self):
         # create_p2p_logfile is taken from slips.yaml
         if self.create_p2p_logfile:
             # rotates p2p.log file every 1 day
-            self.rotator_thread.start()
+            utils.start_thread(self.rotator_thread, self.db)
 
         # should call self.update_callback
         # self.c4 = self.db.subscribe(self.slips_update_channel)