Changed the is_msg_intended_for

Author: A-atmos

conftest.py

@@ -5,7 +5,7 @@
 import pytest
 import os, sys, inspect
 from multiprocessing import Queue
-
+import uuid
 
 # add parent dir to path for imports to work
 current_dir = os.path.dirname(
@@ -47,7 +47,7 @@ def profilerQueue():
 @pytest.fixture
 def database(outputQueue):
     from slips_files.core.database.database import __database__
-    __database__.start(1234)
+    __database__.start(str(uuid.uuid4()))
     __database__.outputqueue = outputQueue
     __database__.print = do_nothing
     return __database__

process_manager.py

@@ -293,7 +293,7 @@ def shutdown_gracefully(self):
                         if message and message['data'] in ('stop_process', 'stop_slips'):
                             continue
 
-                        if utils.is_msg_intended_for(message, 'finished_modules'):
+                        if __database__.is_msg_intended_for(message, 'finished_modules'):
                             # all modules must reply with their names in this channel after
                             # receiving the stop_process msg
                             # to confirm that all processing is done and we can safely exit now

redis_manager.py

@@ -108,15 +108,28 @@ def close_all_instance(self):
         unique_ports = set()
         # close all ports in logfile
         for _id in self.open_servers_IDs:
-            self.flush_redis_id(_id=_id, port = self.open_servers_IDs[_id])
+            # self.flush_redis_id(_id=_id, port = self.open_servers_IDs[_id])
             unique_ports.add(self.open_servers_IDs[_id])
 
         # Kill default port
-
+        
         for port in unique_ports:
             # Skip cache db which is in port 6379
+            r = redis.StrictRedis(
+                host='localhost',
+                port=self.open_servers_IDs[_id],
+                db=0,
+                charset='utf-8',
+                socket_keepalive=True,
+                decode_responses=True,
+                retry_on_timeout=True,
+                health_check_interval=20,
+                )
+            r.flushdb()
+            
             if str(port) == '6379':
                 continue
+            
             pid = self.get_pid_of_redis_server(port = port)
             self.kill_redis_server(pid)
 

slips.py

@@ -744,7 +744,7 @@ def sig_handler(sig, frame):
                 message = self.c1.get_message(timeout=0.01)
                 if (
                     message
-                    and utils.is_msg_intended_for(message, 'finished_modules')
+                    and __database__.is_msg_intended_for(message, 'finished_modules')
                     and message['data'] == 'stop_slips'
                 ):
                     self.proc_man.shutdown_gracefully()

slips_files/common/slips_utils.py

@@ -343,19 +343,6 @@ def get_hash_from_file(self, filename):
                 fb = f.read(BLOCK_SIZE)
         return file_hash.hexdigest()
 
-    def is_msg_intended_for(self, message, channel):
-        """
-        Function to check
-            1. If the given message is intended for this channel
-            2. The msg has valid data
-        """
-
-        return (
-            message
-            and type(message['data']) == str
-            and message['data'] != 'stop_process'
-            and message['channel'] == channel
-        )
 
     def get_branch_info(self):
         """

slips_files/core/database/_profile_flow.py

@@ -1350,7 +1350,7 @@ def get_dns_resolution(self, ip):
         If not resolved, returns {}
         this function is called for every IP in the timeline of kalipso
         """
-        if ip_info := self.r.hget('DNSresolution', ip):
+        if ip_info := self.r.hget(self.prefix + self.separator + 'DNSresolution', ip):
             ip_info = json.loads(ip_info)
             # return a dict with 'ts' 'uid' 'domains' about this IP
             return ip_info

slips_files/core/database/database.py

@@ -358,14 +358,14 @@ def addProfile(self, profileid, starttime, duration):
         """
         try:
             # make sure we don't add public ips if the user specified a home_network
-            if self.r.sismember(self.prefix + self.separator + 'profiles', str(profileid)):
+            if self.r.sismember(self.prefix + self.separator + 'profiles', self.prefix + self.separator + str(profileid)):
                 # we already have this profile
                 return False
             # execlude ips outside of local network is it's set in slips.conf
             if not self.should_add(profileid):
                 return False
             # Add the profile to the index. The index is called prefix + separator + 'profiles'
-            self.r.sadd(self.prefix + self.separator + 'profiles', str(profileid))
+            self.r.sadd(self.prefix + self.separator + 'profiles', self.prefix + self.separator + str(profileid))
             # Create the hashmap with the profileid. The hasmap of each profile is named with the profileid
             # Add the start time of profile
             self.r.hset(self.prefix + self.separator + str(profileid), 'starttime', starttime)
@@ -725,7 +725,7 @@ def getProfileIdFromIP(self, daddr_as_obj):
         """Receive an IP and we want the profileid"""
         try:
             profileid = f'profile{self.separator}{str(daddr_as_obj)}'
-            if data := self.r.sismember('profiles', profileid):
+            if data := self.r.sismember(self.prefix + self.separator + 'profiles', self.prefix + self.separator + str(profileid)):
                 return profileid
             return False
         except redis.exceptions.ResponseError as inst:
@@ -799,7 +799,7 @@ def getT2ForProfileTW(self, profileid, twid, tupleid, tuple_key: str):
 
     def has_profile(self, profileid):
         """Check if we have the given profile"""
-        return self.r.sismember(self.prefix + self.separator + 'profiles', profileid) if profileid else False
+        return self.r.sismember(self.prefix + self.separator + 'profiles', self.prefix + self.separator + str(profileid)) if profileid else False
 
     def getProfilesLen(self):
         """Return the amount of profiles. Redis should be faster than python to do this count"""
@@ -2607,4 +2607,17 @@ def store_std_file(self, **kwargs):
     def get_stdfile(self, file_type):
         return self.r.get(self.prefix + self.separator + str(file_type))
 
+    def is_msg_intended_for(self, message, channel):
+        """
+        Function to check
+            1. If the given message is intended for this channel
+            2. The msg has valid data
+        """
+
+        return (
+            message
+            and type(message['data']) == str
+            and message['data'] != 'stop_process'
+            and message['channel'] == self.prefix + self.separator + str(channel)
+        )
 __database__ = Database()

slips_files/core/evidenceProcess.py

@@ -576,7 +576,7 @@ def run(self):
         while True:
             try:
                 message = __database__.get_message(self.c1)
-                if utils.is_msg_intended_for(message, 'evidence_added'):
+                if __database__.is_msg_intended_for(message, 'evidence_added'):
                     # Data sent in the channel as a json dict, it needs to be deserialized first
                     data = json.loads(message['data'])
                     profileid = data.get('profileid')
@@ -742,7 +742,7 @@ def run(self):
                             )
 
                 message = __database__.get_message(self.c2)
-                if utils.is_msg_intended_for(message, 'new_blame'):
+                if __database__.is_msg_intended_for(message, 'new_blame'):
                     data = message['data']
                     try:
                         data = json.loads(data)

slips_files/core/inputProcess.py

@@ -711,7 +711,7 @@ def remove_old_zeek_files(self):
             msg = __database__.get_message(self.c1)
             if msg and msg['data'] == 'stop_process':
                 return True
-            if utils.is_msg_intended_for(msg, 'remove_old_files'):
+            if __database__.is_msg_intended_for(msg, 'remove_old_files'):
                 # this channel receives renamed zeek log files, we can safely delete them and close their handle
                 changed_files = json.loads(msg['data'])
 

slips_files/core/profilerProcess.py

@@ -2461,7 +2461,7 @@ def run(self):
                 if message and message['data'] == 'stop_process':
                     self.shutdown_gracefully()
                     return True
-                if utils.is_msg_intended_for(message, 'reload_whitelist'):
+                if __database__.is_msg_intended_for(message, 'reload_whitelist'):
                     # if whitelist.conf is edited using pycharm
                     # a msg will be sent to this channel on every keypress, because pycharm saves file automatically
                     # otherwise this channel will get a msg only when whitelist.conf is modified and saved to disk

tests/integration_tests/test_config_files.py

@@ -78,18 +78,18 @@ def check_for_text(txt, output_dir):
     return False
 
 @pytest.mark.parametrize(
-    'pcap_path, expected_profiles, output_dir, redis_port',
+    'pcap_path, expected_profiles, output_dir, prefix',
     [
         (
             'dataset/test7-malicious.pcap',
             290,
             'test_configuration_file/',
-            6667,
+            '2f168df6-c2a9-4a0a-935a-b04fe92e43b7',
         )
     ],
 )
 def test_conf_file(
-    pcap_path, expected_profiles, output_dir, redis_port
+    pcap_path, expected_profiles, output_dir, prefix
 ):
     """
     In this test we're using tests/test.conf
@@ -101,14 +101,16 @@ def test_conf_file(
               f'-f {pcap_path} ' \
               f'-o {output_dir} ' \
               f'-c tests/integration_tests/test.conf  ' \
-              f'-P {redis_port} ' \
+              f'-uid {prefix} ' \
               f'> {output_file} 2>&1'
     # this function returns when slips is done
     os.system(command)
 
     assert has_errors(output_dir) is False
 
-    database = connect_to_redis(redis_port)
+    database = connect_to_redis(6379)
+    database.setPrefix(prefix)
+
     profiles = int(database.getProfilesLen())
     # expected_profiles is more than 50 because we're using direction = all
     assert profiles > expected_profiles
@@ -146,18 +148,18 @@ def test_conf_file(
 
 
 @pytest.mark.parametrize(
-    'pcap_path, expected_profiles, output_dir, redis_port',
+    'pcap_path, expected_profiles, output_dir, prefix',
     [
         (
             'dataset/test8-malicious.pcap',
             1,
             'pcap_test_conf2/',
-            6668,
+            '5eade174-9e34-431b-86c7-4569e55a723d',
         )
     ],
 )
 def test_conf_file2(
-    pcap_path, expected_profiles, output_dir, redis_port
+    pcap_path, expected_profiles, output_dir, prefix
 ):
     """
     In this test we're using tests/test2.conf
@@ -170,15 +172,15 @@ def test_conf_file2(
               f'-f {pcap_path} ' \
               f'-o {output_dir} ' \
               f'-c tests/integration_tests/test2.conf ' \
-              f'-P {redis_port} ' \
+              f'-uid {prefix} ' \
               f'> {output_file} 2>&1'
     # this function returns when slips is done
     os.system(command)
 
     assert has_errors(output_dir) is False
 
-    database = connect_to_redis(redis_port)
-
+    database = connect_to_redis(6379)
+    database.setPrefix(prefix)
     # test 1 homenet ip
     # the only profile we should have is the one in home_network parameter
     profiles = int(database.getProfilesLen())