Analyzer.py - master - KuzminVD/Dumpy

Analyzer.py
728 строк41 KB
vladimir
Load code
03 июн 2024, 00:08
03 июн 2024, 00:08f404d0f
import jellyfish
import pandas as pd
from colorama import Fore
import msticpy as mp
from msticpy.vis import timeline_duration
from msticpy.vis.matrix_plot import plot_matrix
from msticpy.transform.iocextract import IoCExtract
from bokeh.resources import CDN
from bokeh.embed import file_html


class Analyzer():
    def __init__(self, plugs_dir, output_dir):
        self.plugs_dir = plugs_dir
        self.output_dir = output_dir
        self.ti = mp.TILookup()  # ./msticpyconfig.yaml
        self.dfprocs = pd.read_csv(plugs_dir + 'windows.pslist.PsList.csv')
        self.dfcmdline = pd.read_csv(plugs_dir + 'windows.cmdline.CmdLine.csv')
        self.dfnetscan = pd.read_csv(plugs_dir + 'windows.netscan.NetScan.csv')
        self.dfcallbacks = pd.read_csv(plugs_dir + 'windows.callbacks.Callbacks.csv')
        self.dfdriverscan = pd.read_csv(plugs_dir + 'windows.driverscan.DriverScan.csv')
        self.dfdlllist = pd.read_csv(plugs_dir + 'windows.dlllist.DllList.csv')
        self.dfsvcscan = pd.read_csv(plugs_dir + 'windows.svcscan.SvcScan.csv')
        self.dfhandles = pd.read_csv(plugs_dir + 'windows.handles.Handles.csv')
        self.dfmalfind = pd.read_csv(plugs_dir + 'windows.malfind.Malfind.csv')
        self.dfssdt = pd.read_csv(plugs_dir + 'windows.ssdt.SSDT.csv')
        self.dfldrmod = pd.read_csv(plugs_dir + 'windows.ldrmodules.LdrModules.csv')
        self.dfppidtxt = self.dfprocs[['PID', 'ImageFileName']].drop_duplicates()
        self.dfppidtxt.columns = ['PPID', 'ParentName']
        self.dfprocs = self.dfprocs.merge(self.dfppidtxt, on='PPID', how='left')

    def Test_Proc_1(self):
        output = pd.DataFrame(columns=['script', 'message', 'code'])
        message = "Identifying processes that should not have more than one instance"
        print(Fore.YELLOW + message)
        output.loc[len(output.index)] = [self.Test_Proc_1.__name__, message, 'info_title']
        status = 0
        singleprocs = ["system",
                       "wininit.exe",
                       "lsass.exe",
                       "lsm.exe",
                       "services.exe",
                       "lsaiso.exe"]
        for procs in singleprocs:
            p_count = self.dfprocs[self.dfprocs.ImageFileName.str.lower() == procs].shape[0]
            if p_count == 1:
                message = "No multiple instance(s) of " + procs + " found"
                print(Fore.GREEN + message)
                output.loc[len(output.index)] = [self.Test_Proc_1.__name__, message, 'info_good']
            elif p_count > 1:
                message = "Multiple instance(s) of " + procs + " found!"
                print(Fore.RED + message)
                output.loc[len(output.index)] = [self.Test_Proc_1.__name__, message, 'info_bad']
                message = self.dfprocs[self.dfprocs.ImageFileName.str.lower() == procs][
                          ['ImageFileName', 'PID', 'PPID', 'ParentName']].to_json(orient='records')
                print(message)
                output.loc[len(output.index)] = [self.Test_Proc_1.__name__, message, 'info_detect']
                status += 1
        return status, output
    def Test_Proc_2(self):
        output = pd.DataFrame(columns=['script', 'message', 'code'])
        message = "Identifying suspicious parent-child process lineage"
        print(Fore.YELLOW + message)
        output.loc[len(output.index)] = [self.Test_Proc_2.__name__, message, 'info_title']
        status = 0
        sus_services = \
        self.dfprocs[(self.dfprocs.ImageFileName.str.lower() == "services.exe") & (self.dfprocs.ParentName.str.lower() != "wininit.exe")][
            ['ImageFileName', 'PID', 'PPID', 'ParentName']]
        if (sus_services.empty):
            message = "No suspicious instances of services.exe were found"
            print(Fore.GREEN + message)
            output.loc[len(output.index)] = [self.Test_Proc_2.__name__, message, 'info_good']
        else:
            message = "Suspicious services.exe process(s) found"
            print(Fore.RED + message)
            output.loc[len(output.index)] = [self.Test_Proc_2.__name__, message, 'info_bad']
            message = sus_services.to_json(orient='records')
            print(message)
            output.loc[len(output.index)] = [self.Test_Proc_2.__name__, message, 'info_detect']
            status += 1

        sus_iexplorer = \
        self.dfprocs[(self.dfprocs.ImageFileName.str.lower() == "iexplore.exe") & (self.dfprocs.ParentName.str.lower() != "explorer.exe")][
            ['ImageFileName', 'PID', 'PPID', 'ParentName']]
        if (sus_iexplorer.empty):
            message = "No suspicious instances of iexplore.exe were found"
            print(Fore.GREEN + message)
            output.loc[len(output.index)] = [self.Test_Proc_2.__name__, message, 'info_good']
        else:
            message = "iexplore.exe spawned from a suspicious parent"
            print(Fore.RED + message)
            output.loc[len(output.index)] = [self.Test_Proc_2.__name__, message, 'info_bad']
            message = sus_iexplorer.to_json(orient='records')
            print(message)
            output.loc[len(output.index)] = [self.Test_Proc_2.__name__, message, 'info_detect']
            status += 1

        sus_svchost = \
        self.dfprocs[(self.dfprocs.ImageFileName.str.lower() == "svchost.exe") & (self.dfprocs.ParentName.str.lower() != "services.exe")][
            ['ImageFileName', 'PID', 'PPID', 'ParentName']]
        if (sus_svchost.empty):
            message = "No suspicious instances of svchost.exe were found"
            print(Fore.GREEN + message)
            output.loc[len(output.index)] = [self.Test_Proc_2.__name__, message, 'info_good']
        else:
            message = "svchost.exe spawned from a suspicious parent"
            print(Fore.RED + message)
            output.loc[len(output.index)] = [self.Test_Proc_2.__name__, message, 'info_bad']
            message = sus_svchost.to_json(orient='records')
            print(message)
            output.loc[len(output.index)] = [self.Test_Proc_2.__name__, message, 'info_detect']
            status += 1

        sus_cli = self.dfprocs[
            ((self.dfprocs.ImageFileName.str.lower() == "cmd.exe") | (self.dfprocs.ImageFileName.str.lower() == "powershell.exe")) & (
                        self.dfprocs.ParentName.str.lower() != "explorer.exe")][['ImageFileName', 'PID', 'PPID', 'ParentName']]
        if (sus_cli.empty):
            message = "No suspicious instances of cmd.exe or powershell.exe were found"
            print(Fore.GREEN + message)
            output.loc[len(output.index)] = [self.Test_Proc_2.__name__, message, 'info_good']
        else:
            message = "cmd.exe or powershell.exe spawned by an unusual parent!"
            print(Fore.RED + message)
            output.loc[len(output.index)] = [self.Test_Proc_2.__name__, message, 'info_bad']
            message = sus_cli.to_json(orient='records')
            print(message)
            output.loc[len(output.index)] = [self.Test_Proc_2.__name__, message, 'info_detect']
            status += 1

        return status, output
    def Test_Proc_3(self):
        output = pd.DataFrame(columns=['script', 'message', 'code'])
        message = "Identifying suspicious processes using Damerau-Levenshtein distance"
        print(Fore.YELLOW + message)
        output.loc[len(output.index)] = [self.Test_Proc_3.__name__, message, 'info_title']
        status = 0
        actprocsbaselst = ["svchost.exe",
                           "wininit.exe",
                           "lsass.exe",
                           "explorer.exe",
                           "iexplore.exe",
                           "powershell.exe",
                           "cmd.exe",
                           "services.exe",
                           "csrss.exe",
                           "smss.exe",
                           "silhost.exe",
                           "taskhost.exe",
                           "MpCmdRun.exe"]
        liveprocs = []
        levratedprocs = []
        for susproc in self.dfprocs["ImageFileName"].str.lower():
            if susproc not in liveprocs:
                liveprocs.append(susproc)

        for x, y in [(x, y) for x in liveprocs for y in actprocsbaselst]:
            levdistance = (jellyfish.damerau_levenshtein_distance(x, y))
            if 0 < levdistance <= 2:
                levratedprocs.append([x, y, levdistance])

        dflevprocs = pd.DataFrame(levratedprocs, columns=['LiveProcess', 'BaseProcess', 'Damerau–Levenshtein distance'])
        procfpredux = dflevprocs[((dflevprocs.LiveProcess.str.lower().isin(
            ['lsass.exe', 'csrss.exe', 'smss.exe', 'dllhost.exe', 'explorer.exe', 'svchost.exe', 'sihost.exe'])) & (
                                      dflevprocs.BaseProcess.str.lower().isin(
                                          ['lsass.exe', 'csrss.exe', 'smss.exe', 'silhost.exe', 'iexplore.exe',
                                           'svchost.exe']))) == False]
        if (procfpredux.empty):
            message = "No processes were identified for Damerau-Levenshtein analysis"
            print(Fore.GREEN + message)
            output.loc[len(output.index)] = [self.Test_Proc_3.__name__, message, 'info_good']
        else:
            message = "Suspicious processes were identified"
            print(Fore.RED + message)
            output.loc[len(output.index)] = [self.Test_Proc_3.__name__, message, 'info_bad']
            message = procfpredux.to_json(orient='records')
            print(message)
            output.loc[len(output.index)] = [self.Test_Proc_3.__name__, message, 'info_detect']
            status += 1

        return status, output
    def Test_Proc_4(self):
        output = pd.DataFrame(columns=['script', 'message', 'code'])
        message = "Identifying svchost.exe or rundll32.exe without any command line arguments"
        print(Fore.YELLOW + message)
        output.loc[len(output.index)] = [self.Test_Proc_4.__name__, message, 'info_title']
        status = 0
        nocmdargs_svc = (self.dfcmdline[self.dfcmdline["Args"].str.match(r'^(.*(svchost|rundll32)\.exe)$')])[
            ['PID', 'Process', 'Args']]
        if (nocmdargs_svc.empty):
            message = "No svchost.exe or rundll32.exe instance(s) identified without commandline arguments"
            print(Fore.GREEN + message)
            output.loc[len(output.index)] = [self.Test_Proc_4.__name__, message, 'info_good']
        else:
            message = "svchost.exe or rundll32.exe without command line was detected!"
            print(Fore.RED + message)
            output.loc[len(output.index)] = [self.Test_Proc_4.__name__, message, 'info_bad']
            message = nocmdargs_svc.to_json(orient='records')
            print(message)
            output.loc[len(output.index)] = [self.Test_Proc_4.__name__, message, 'info_detect']
            status += 1

        return status, output
    def Test_Proc_5(self):
        output = pd.DataFrame(columns=['script', 'message', 'code'])
        message = "Executables having paths in suspicious folder locations"
        print(Fore.YELLOW + message)
        output.loc[len(output.index)] = [self.Test_Proc_5.__name__, message, 'info_title']
        status = 0
        nocmdargs_svc = (self.dfcmdline[self.dfcmdline["Args"].str.match(r'^(.*(svchost|rundll32)\.exe)$')])[
            ['PID', 'Process', 'Args']]
        if (nocmdargs_svc.empty):
            message = "No svchost.exe or rundll32.exe instance(s) identified without commandline arguments"
            print(Fore.GREEN + message)
            output.loc[len(output.index)] = [self.Test_Proc_5.__name__, message, 'info_good']
        else:
            message = "svchost.exe or rundll32.exe without command line was detected!"
            print(Fore.RED + message)
            output.loc[len(output.index)] = [self.Test_Proc_5.__name__, message, 'info_bad']
            message = nocmdargs_svc.to_json(orient='records')
            print(message)
            output.loc[len(output.index)] = [self.Test_Proc_5.__name__, message, 'info_detect']
            status += 1

        return status, output
    def Test_Proc_6(self):
        output = pd.DataFrame(columns=['script', 'message', 'code'])
        message = "Process network graph"
        print(Fore.YELLOW + message)
        output.loc[len(output.index)] = [self.Test_Proc_6.__name__, message, 'info_title']
        status = 0
        graph = self.dfprocs.mp_plot.network(
            source_col="ImageFileName",
            target_col="ParentName",
            source_attrs=["PID", "ParentName", "CreateTime", "ExitTime"],
            edge_attrs=["CreateTime"],
            graph_type="digraph"
        )
        save_dir = self.output_dir + self.Test_Proc_6.__name__ + '.html'
        html = file_html(graph, CDN, self.Test_Proc_6.__name__)
        with open(save_dir, 'w') as f:
            f.write(html)
        message = "Process network graph was saved to: " + save_dir
        output.loc[len(output.index)] = [self.Test_Proc_6.__name__, message, 'info_info']

        return status, output
    def Test_Proc_7(self):
        output = pd.DataFrame(columns=['script', 'message', 'code'])
        message = "Anomalous Process Detection using time duration"
        print(Fore.YELLOW + message)
        output.loc[len(output.index)] = [self.Test_Proc_7.__name__, message, 'info_title']
        status = 0
        dfprocs_exit = self.dfprocs.fillna('')
        graph = timeline_duration.display_timeline_duration(
            data=dfprocs_exit,
            group_by="ImageFileName",
            time_column="CreateTime",
            end_time_column="ExitTime"
        )
        save_dir = self.output_dir + self.Test_Proc_7.__name__ + '.html'
        html = file_html(graph, CDN, self.Test_Proc_7.__name__)
        with open(save_dir, 'w') as f:
            f.write(html)
        message = "Anomalous Process Detection using time duration was saved to: " + save_dir
        output.loc[len(output.index)] = [self.Test_Proc_7.__name__, message, 'info_info']
        return status, output
    def Test_Proc_8(self):
        output = pd.DataFrame(columns=['script', 'message', 'code'])
        message = "Suspicious commandline arguments"
        print(Fore.YELLOW + message)
        output.loc[len(output.index)] = [self.Test_Proc_8.__name__, message, 'info_title']
        status = 0
        sus_cmdargs = self.dfcmdline[self.dfcmdline["Args"].str.match(
            r".*(nop|NoLogo|noni|invoke-expression|WebClient|EncodedCommand|nop|reflection\.assembly|-enc|IO\.Compression|DownloadString|System\.Net\.WebClient|Hidden|iex|FromBase64|powershell\.cpl|base64|MemoryStream)|COMSPEC|IABz|SQBFAFgA|aWV4I|aQBlAHgA|SUVYI|sekurlsa|(whoami|certutil|bitsadmin|mshta|StartW|MiniDump|comsvcs\.dll).*",
            case=False) == True][['PID', 'Process', 'Args']]
        if (sus_cmdargs.empty):
            message = "No suspicious commandline arguments detected"
            print(Fore.GREEN + message)
            output.loc[len(output.index)] = [self.Test_Proc_8.__name__, message, 'info_good']
        else:
            message = "Suspicious commandline arguments detected!"
            print(Fore.RED + message)
            output.loc[len(output.index)] = [self.Test_Proc_8.__name__, message, 'info_bad']
            message = sus_cmdargs.to_json(orient='records')
            print(message)
            output.loc[len(output.index)] = [self.Test_Proc_8.__name__, message, 'info_detect']
            status += 1

        ioc_extractor = IoCExtract()
        ioc_df = ioc_extractor.extract(data=self.dfcmdline, columns=['Args'])
        if len(ioc_df):
            message = "IoC patterns found in process tree"
            print(message)
            output.loc[len(output.index)] = [self.Test_Proc_8.__name__, message, 'info_bad']
            message = ioc_df.to_json(orient='records')
            print(message)
            output.loc[len(output.index)] = [self.Test_Proc_8.__name__, message, 'info_detect']
            status += 1

        return status, output
    def Test_Proc_9(self):
        output = pd.DataFrame(columns=['script', 'message', 'code'])
        message = "rundll32.exe spawned by a suspicious parent process"
        print(Fore.YELLOW + message)
        output.loc[len(output.index)] = [self.Test_Proc_9.__name__, message, 'info_title']
        status = 0
        susdllloc = self.dfprocs[((self.dfprocs.ImageFileName.str.lower() == "rundll32.exe")) & (self.dfprocs.ParentName.str.lower().isin(
            ['winword.exe', 'excel.exe', 'msaccess.exe', 'lsass.exe', 'taskeng.exe', 'winlogon.exe', 'schtask.exe',
             'regsvr32.exe', 'wmiprvse.exe', 'wsmprovhost.exe']))][['ImageFileName', 'PID', 'PPID', 'ParentName']]
        if (susdllloc.empty):
            message = "No suspicious rundll32 parent detected"
            print(Fore.GREEN + message)
            output.loc[len(output.index)] = [self.Test_Proc_9.__name__, message, 'info_good']
        else:
            message = "Suspicious rundll32.exe parent detected!"
            print(Fore.RED + message)
            output.loc[len(output.index)] = [self.Test_Proc_9.__name__, message, 'info_bad']
            message = susdllloc.to_json(orient='records')
            print(message)
            output.loc[len(output.index)] = [self.Test_Proc_9.__name__, message, 'info_detect']
            status += 1

        return status, output
    def Test_Net_1(self):
        output = pd.DataFrame(columns=['script', 'message', 'code'])
        message = "Visual network interactions"
        print(Fore.YELLOW + message)
        output.loc[len(output.index)] = [self.Test_Net_1.__name__, message, 'info_title']
        status = 0
        matrix = plot_matrix(data=self.dfnetscan, x="ForeignAddr", y="LocalAddr", title="IP interactions", sort="dsc", intersect=True)
        save_dir = self.output_dir + self.Test_Net_1.__name__ + '.html'
        html = file_html(matrix, CDN, self.Test_Net_1.__name__)
        with open(save_dir, 'w') as f:
            f.write(html)
        message = "Visual network interactions was saved to: " + save_dir
        output.loc[len(output.index)] = [self.Test_Net_1.__name__, message, 'info_info']
        return status, output
    def Test_Net_2(self):
        output = pd.DataFrame(columns=['script', 'message', 'code'])
        message = "No of external network communications per public IP"
        print(Fore.YELLOW + message)
        output.loc[len(output.index)] = [self.Test_Net_2.__name__, message, 'info_title']
        status = 0
        netscans = self.dfnetscan[(self.dfnetscan["ForeignAddr"].str.match(
            r'(?!^0\.)(?!^10\.)(?!^100\.6[4-9]\.)(?!^100\.[7-9]\d\.)(?!^100\.1[0-1]\d\.)(?!^100\.12[0-7]\.)(?!^127\.)(?!^169\.254\.)(?!^172\.1[6-9]\.)(?!^172\.2[0-9]\.)(?!^172\.3[0-1]\.)(?!^192\.0\.0\.)(?!^192\.0\.2\.)(?!^192\.88\.99\.)(?!^192\.168\.)(?!^198\.1[8-9]\.)(?!^198\.51\.100\.)(?!^203.0\.113\.)(?!^22[4-9]\.)(?!^23[0-9]\.)(?!^24[0-9]\.)(?!^25[0-5]\.)(([0-9]|[1-9][0-9]|1[0-9]{2}|2[0-4][0-9]|25[0-5])\.([0-9]|[1-9][0-9]|1[0-9]{2}|2[0-4][0-9]|25[0-5])\.([0-9]|[1-9][0-9]|1[0-9]{2}|2[0-4][0-9]|25[0-5])\.([0-9]|[1-9][0-9]|1[0-9]{2}|2[0-4][0-9]|25[0-5]))$') == True)][
            "ForeignAddr"]
        if (netscans.empty):
            message = "No of external network communications per public IP = 0"
            output.loc[len(output.index)] = [self.Test_Net_2.__name__, message, 'info_info']
        else:
            ax = netscans.value_counts().plot(kind='bar')
            ax.tick_params(axis='x', rotation=0)
            fig = ax.get_figure()
            save_dir = self.output_dir + self.Test_Net_2.__name__ + '.png'
            fig.savefig(save_dir)
            fig.show()
            message = "No of external network communications per public IP was saved to: " + save_dir
            output.loc[len(output.index)] = [self.Test_Net_2.__name__, message, 'info_info']

        return status, output
    def Test_Net_3(self):
        output = pd.DataFrame(columns=['script', 'message', 'code'])
        message = "Intelligence matches for external network connections"
        print(Fore.YELLOW + message)
        output.loc[len(output.index)] = [self.Test_Net_3.__name__, message, 'info_title']
        status = 0
        ext_netclose = self.dfnetscan[(self.dfnetscan["ForeignAddr"].str.match(
            r'(?!^0\.)(?!^10\.)(?!^100\.6[4-9]\.)(?!^100\.[7-9]\d\.)(?!^100\.1[0-1]\d\.)(?!^100\.12[0-7]\.)(?!^127\.)(?!^169\.254\.)(?!^172\.1[6-9]\.)(?!^172\.2[0-9]\.)(?!^172\.3[0-1]\.)(?!^192\.0\.0\.)(?!^192\.0\.2\.)(?!^192\.88\.99\.)(?!^192\.168\.)(?!^198\.1[8-9]\.)(?!^198\.51\.100\.)(?!^203.0\.113\.)(?!^22[4-9]\.)(?!^23[0-9]\.)(?!^24[0-9]\.)(?!^25[0-5]\.)(([0-9]|[1-9][0-9]|1[0-9]{2}|2[0-4][0-9]|25[0-5])\.([0-9]|[1-9][0-9]|1[0-9]{2}|2[0-4][0-9]|25[0-5])\.([0-9]|[1-9][0-9]|1[0-9]{2}|2[0-4][0-9]|25[0-5])\.([0-9]|[1-9][0-9]|1[0-9]{2}|2[0-4][0-9]|25[0-5]))$') == True)][
            ["ForeignAddr", "ForeignPort", "Owner"]]
        if (ext_netclose.empty):
            message = "No closed external connections were detected"
            print(Fore.GREEN + message)
            output.loc[len(output.index)] = [self.Test_Net_3.__name__, message, 'info_good']
        else:
            message = "Terminated external network connections detected"
            print(Fore.RED + message)
            output.loc[len(output.index)] = [self.Test_Net_3.__name__, message, 'info_bad']
            message = ext_netclose.to_json(orient='records')
            print(message)
            output.loc[len(output.index)] = [self.Test_Net_3.__name__, message, 'info_detect']
            print()
            message = "Positive Threat Intelligence Enrichment on the IPs"
            print(message)
            output.loc[len(output.index)] = [self.Test_Net_3.__name__, message, 'info_info']
            ext_netclose.drop_duplicates(subset="ForeignAddr", keep=False, inplace=True)
            resraw_intel = self.ti.lookup_iocs(data=ext_netclose, ioc_col="ForeignAddr")
            res_intel = resraw_intel[resraw_intel["Result"] == True]
            res_intel_cut = res_intel.drop(['QuerySubtype', 'RawResult', 'Result'], axis=1)
            message = res_intel_cut.to_json(orient='records')
            print(message)
            output.loc[len(output.index)] = [self.Test_Net_3.__name__, message, 'info_detect']
            status += 1

        return status, output
    def Test_Net_4(self):
        output = pd.DataFrame(columns=['script', 'message', 'code'])
        message = "ISuspicious web connections that are not related to browsers"
        print(Fore.YELLOW + message)
        output.loc[len(output.index)] = [self.Test_Net_4.__name__, message, 'info_title']
        status = 0
        sus_http = self.dfnetscan[
            (self.dfnetscan["ForeignPort"] == 8080) | (self.dfnetscan["ForeignPort"] == 80) | (self.dfnetscan["ForeignPort"] == 443) & (
                        (self.dfnetscan.Owner.str.lower() != "msmpeng.exe") & (self.dfnetscan.Owner.str.lower() != "chrome.exe") & (
                            self.dfnetscan.Owner.str.lower() != "msedge.exe") & (
                                    self.dfnetscan.Owner.str.lower() != "firefox.exe") & (
                                    self.dfnetscan.Owner.str.lower() != "iexplorer.exe"))][
            ["ForeignAddr", "ForeignPort", "Owner"]]
        if (sus_http.empty):
            message = "No suspicious web connections were detected"
            print(Fore.GREEN + message)
            output.loc[len(output.index)] = [self.Test_Net_4.__name__, message, 'info_good']
        else:
            message = "None browser based web connections were detected"
            print(Fore.RED + message)
            output.loc[len(output.index)] = [self.Test_Net_4.__name__, message, 'info_bad']
            message = sus_http.to_json(orient='records')
            print(message)
            output.loc[len(output.index)] = [self.Test_Net_4.__name__, message, 'info_detect']
            print()
            message = "Positive Threat Intelligence Enrichment on the IPs"
            print(message)
            output.loc[len(output.index)] = [self.Test_Net_4.__name__, message, 'info_info']
            resraw_intel2 = self.ti.lookup_iocs(data=sus_http, ioc_col="ForeignAddr", providers=["GreyNoise", "OTX"])
            res_intel2 = resraw_intel2[resraw_intel2["Result"] == True]
            res_intel2_cut = res_intel2.drop(['QuerySubtype', 'RawResult', 'Result'], axis=1)
            if (res_intel2_cut.empty):
                message = "No Enrichments"
                print(message)
                output.loc[len(output.index)] = [self.Test_Net_4.__name__, message, 'info_good']
            else:
                message = "Found Enrichments"
                print(Fore.RED + message)
                output.loc[len(output.index)] = [self.Test_Net_4.__name__, message, 'info_bad']
                message = res_intel2_cut.to_json(orient='records')
                print(message)
                output.loc[len(output.index)] = [self.Test_Net_4.__name__, message, 'info_detect']
            status += 1

        return status, output
    def Test_Net_5(self):
        output = pd.DataFrame(columns=['script', 'message', 'code'])
        message = "IP address in commandline arguments"
        print(Fore.YELLOW + message)
        output.loc[len(output.index)] = [self.Test_Net_5.__name__, message, 'info_title']
        status = 0
        ip_cmdline = self.dfcmdline[self.dfcmdline["Args"].str.match(
            r'.*(?!^0\.)(?!^10\.)(?!^100\.6[4-9]\.)(?!^100\.[7-9]\d\.)(?!^100\.1[0-1]\d\.)(?!^100\.12[0-7]\.)(?!^127\.)(?!^169\.254\.)(?!^172\.1[6-9]\.)(?!^172\.2[0-9]\.)(?!^172\.3[0-1]\.)(?!^192\.0\.0\.)(?!^192\.0\.2\.)(?!^192\.88\.99\.)(?!^192\.168\.)(?!^198\.1[8-9]\.)(?!^198\.51\.100\.)(?!^203.0\.113\.)(?!^22[4-9]\.)(?!^23[0-9]\.)(?!^24[0-9]\.)(?!^25[0-5]\.)(([0-9]|[1-9][0-9]|1[0-9]{2}|2[0-4][0-9]|25[0-5])\.([0-9]|[1-9][0-9]|1[0-9]{2}|2[0-4][0-9]|25[0-5])\.([0-9]|[1-9][0-9]|1[0-9]{2}|2[0-4][0-9]|25[0-5])\.([0-9]|[1-9][0-9]|1[0-9]{2}|2[0-4][0-9]|25[0-5])).*') == True][
            ['PID', 'Process', 'Args']]
        if (ip_cmdline.empty):
            message = "No IPs in commandline arguments detected"
            print(Fore.GREEN + message)
            output.loc[len(output.index)] = [self.Test_Net_5.__name__, message, 'info_good']
        else:
            message = "IP address in commandline arguments detected!"
            print(Fore.RED + message)
            output.loc[len(output.index)] = [self.Test_Net_5.__name__, message, 'info_bad']
            message = ip_cmdline.to_json(orient='records')
            print(message)
            output.loc[len(output.index)] = [self.Test_Net_5.__name__, message, 'info_detect']
            status += 1

        return status, output
    def Test_SysT_1(self):
        output = pd.DataFrame(columns=['script', 'message', 'code'])
        message = "Check for malicious Kernel callbacks (indicator of rootkit activity)"
        print(Fore.YELLOW + message)
        output.loc[len(output.index)] = [self.Test_SysT_1.__name__, message, 'info_title']
        status = 0
        rootkit_indicator = self.dfcallbacks[(self.dfcallbacks.Module.str.lower() == "unknown")][
            ["Type", "Callback", "Module", "Detail"]]
        if (rootkit_indicator.empty):
            message = "No suspicious callbacks detected"
            print(Fore.GREEN + message)
            output.loc[len(output.index)] = [self.Test_SysT_1.__name__, message, 'info_good']
        else:
            message = "Suspicious callbacks identified (behaviour of a rootkit)!"
            print(Fore.RED + message)
            output.loc[len(output.index)] = [self.Test_SysT_1.__name__, message, 'info_bad']
            message = rootkit_indicator.to_json(orient='records')
            print(message)
            output.loc[len(output.index)] = [self.Test_SysT_1.__name__, message, 'info_detect']
            status += 1

        return status, output
    def Test_SysT_2(self):
        output = pd.DataFrame(columns=['script', 'message', 'code'])
        message = "Check for malicious Kernel callbacks (indicator of rootkit activity)"
        print(Fore.YELLOW + message)
        output.loc[len(output.index)] = [self.Test_SysT_2.__name__, message, 'info_title']
        status = 0
        sus_kernelhook = self.dfssdt[(self.dfssdt["Module"].str.match(r".*(ntoskrnl1|win32k).*", case=False) == False)][
            ["Address", "Module", "Symbol"]]
        if (sus_kernelhook.empty):
            message = "No suspicious SSDT hooking detected"
            print(Fore.GREEN + message)
            output.loc[len(output.index)] = [self.Test_SysT_2.__name__, message, 'info_good']
        else:
            message = "Suspicious SSDT hooking detected!"
            print(Fore.RED + message)
            output.loc[len(output.index)] = [self.Test_SysT_2.__name__, message, 'info_bad']
            for row in range(len(sus_kernelhook.index)):
                message = sus_kernelhook[sus_kernelhook.index == row].copy().to_json(orient='records')
                print(message)
                output.loc[len(output.index)] = [self.Test_SysT_2.__name__, message, 'info_detect']
            status += 1

        return status, output
    def Test_SysT_3(self):
        output = pd.DataFrame(columns=['script', 'message', 'code'])
        message = "Check for drivers with the base address wiped out (indicator of rootkit activity)"
        print(Fore.YELLOW + message)
        output.loc[len(output.index)] = [self.Test_SysT_3.__name__, message, 'info_title']
        status = 0
        sus_driveraddwipe = self.dfdriverscan[(self.dfdriverscan.Start.str.lower() == "0x00000000")][
            ["Offset", "Start", "Driver Name", "Name", "Service Key"]]
        if (sus_driveraddwipe.empty):
            message = "No drivers found with zeroed out base address"
            print(Fore.GREEN + message)
            output.loc[len(output.index)] = [self.Test_SysT_3.__name__, message, 'info_good']
        else:
            message = "Driver(s) found with base address zeroed out (Stealth)!"
            print(Fore.RED + message)
            output.loc[len(output.index)] = [self.Test_SysT_3.__name__, message, 'info_bad']
            message = sus_driveraddwipe.to_json(orient='records')
            print(message)
            output.loc[len(output.index)] = [self.Test_SysT_3.__name__, message, 'info_detect']
            status += 1

        return status, output
    def Test_SVC_1(self):
        output = pd.DataFrame(columns=['script', 'message', 'code'])
        message = "Suspicious Windows service file names"
        print(Fore.YELLOW + message)
        output.loc[len(output.index)] = [self.Test_SVC_1.__name__, message, 'info_title']
        status = 0
        sus_svcfnres = self.dfsvcscan[self.dfsvcscan["Binary"].str.match(r".*(cmd|comspec|powershell|echo|%SYSTEMROOT%).*", case=False) == True][
            ['Offset', 'PID', 'Start', 'State', 'Type', 'Name', 'Display', 'Binary']]
        if (sus_svcfnres.empty):
            message = "No suspicious Windows service file name(s) identified"
            print(Fore.GREEN + message)
            output.loc[len(output.index)] = [self.Test_SVC_1.__name__, message, 'info_good']
        else:
            message = "Suspicious Windows service file name(s) identified!"
            print(Fore.RED + message)
            output.loc[len(output.index)] = [self.Test_SVC_1.__name__, message, 'info_bad']
            message = sus_svcfnres.to_json(orient='records')
            print(message)
            output.loc[len(output.index)] = [self.Test_SVC_1.__name__, message, 'info_detect']
            status += 1

        return status, output
    def Test_SVC_2(self):
        output = pd.DataFrame(columns=['script', 'message', 'code'])
        message = "Suspicious Windows Service Names"
        print(Fore.YELLOW + message)
        output.loc[len(output.index)] = [self.Test_SVC_2.__name__, message, 'info_title']
        status = 0
        sussvcnameres = self.dfsvcscan[(self.dfsvcscan["Name"].str.match(
            r"^([a-zA-Z]{4}|[a-zA-Z]{8}|[a-zA-Z]{16}|[a-z0-9]{11}|BTOBTO|WCESERVICE|WCE SERVICE|winexesvc|DumpSvc|pwdump|gsecdump|cachedump)$",
            case=False) == True) & (self.dfsvcscan["Start"] == "SERVICE_BOOT_START") & (self.dfsvcscan["Binary"].str.match(
            r'(?i)(C:\\Windows\\System32\\)|((C:\\ProgramData\\Microsoft)|(C:\\Program Files\\Common Files)|\\Driver.*)',
            na=False, case=False) == False)][['Offset', 'PID', 'Start', 'State', 'Type', 'Name', 'Display', 'Binary']]
        if (sussvcnameres.empty):
            message = "No suspicious Windows service names identified"
            print(Fore.GREEN + message)
            output.loc[len(output.index)] = [self.Test_SVC_2.__name__, message, 'info_good']
        else:
            message = "Suspicious Windows service name(s) identified!"
            print(Fore.RED + message)
            output.loc[len(output.index)] = [self.Test_SVC_2.__name__, message, 'info_bad']
            message = sussvcnameres.to_json(orient='records')
            print(message)
            output.loc[len(output.index)] = [self.Test_SVC_2.__name__, message, 'info_detect']
            status += 1

        return status, output
    def Test_NamedPipe_1(self):
        output = pd.DataFrame(columns=['script', 'message', 'code'])
        message = "Suspicious named pipe(s)"
        print(Fore.YELLOW + message)
        output.loc[len(output.index)] = [self.Test_NamedPipe_1.__name__, message, 'info_title']
        status = 0
        susnamedpipe = self.dfhandles[(self.dfhandles["Type"] == "File") & (self.dfhandles["Name"].str.match(
            r"\\Device\\NamedPipe\\(MSSE-[0-9a-f]{3,4}-server|status_[0-9a-f]{2}|msagent_[0-9a-f]{2}|postex_ssh_[0-9a-f]{4}|postex_[0-9a-f]{4}|mojo\.5688\.8052|wkssvc[0-9a-f]{2}|ntsvcs[0-9a-f]{2}|DserNamePipe[0-9a-f]{2}|SearchTextHarvester[0-9a-f]{2}|mypipe-f[0-9a-f]{2}|mypipe-h[0-9a-f]{2}|windows\.update\.manager[0-9a-f]{2}|windows\.update\.manager[0-9a-f]{3}|ntsvcs_[0-9a-f]{2}|scerpc_[0-9a-f]{2}|scerpc[0-9a-f]{2}|ntsvcs[0-9a-f]{2}|gruntsvc|psexesvc|remcom_comunication|PAExec|csexecsvc|svcctl|Posh|jaccdpqnvbrrxlaf|TestSVC|remcom).*",
            case=False) == True)][['PID', 'Process', 'Offset', 'Type', 'GrantedAccess', 'Name']]
        if (susnamedpipe.empty):
            message = "No suspicious named pipes were identified"
            print(Fore.GREEN + message)
            output.loc[len(output.index)] = [self.Test_NamedPipe_1.__name__, message, 'info_good']
        else:
            message = "Suspicious named pipe(s) detected!"
            print(Fore.RED + message)
            output.loc[len(output.index)] = [self.Test_NamedPipe_1.__name__, message, 'info_bad']
            message = susnamedpipe.to_json(orient='records')
            print(message)
            output.loc[len(output.index)] = [self.Test_NamedPipe_1.__name__, message, 'info_detect']
            status += 1

        return status, output
    def Test_DLL_1(self):
        output = pd.DataFrame(columns=['script', 'message', 'code'])
        message = "DLLs or Sys files located in unusual locations"
        print(Fore.YELLOW + message)
        output.loc[len(output.index)] = [self.Test_DLL_1.__name__, message, 'info_title']
        status = 0
        susdllloc = self.dfdlllist[(self.dfdlllist["Path"].str.match(
            r"^[A-Za-z]:\\.*(Temp|INetCache|Desktop|Downloads|Content|Documents|Public|ProgramData|Program Files|Program Files \(x86\))\\(\w+).(dll|exe)$",
            case=False) == True)][['PID', 'Process', 'Base', 'Name', 'Path', 'LoadTime']]
        if (susdllloc.empty):
            message = "No DLL(s) located in unusual location(s)"
            print(Fore.GREEN + message)
            output.loc[len(output.index)] = [self.Test_DLL_1.__name__, message, 'info_good']
        else:
            message = "Suspicious file identified in an unusual location!"
            print(Fore.RED + message)
            output.loc[len(output.index)] = [self.Test_DLL_1.__name__, message, 'info_bad']
            message = susdllloc.to_json(orient='records')
            print(message)
            output.loc[len(output.index)] = [self.Test_DLL_1.__name__, message, 'info_detect']
            status += 1
            # Add other files associated with this finding for context
            # Add potential mutexes created by the suspicious processe

        return status, output
    def Test_DLL_2(self):
        output = pd.DataFrame(columns=['script', 'message', 'code'])
        message = "DLL(s) or Sys files with unusually long file names"
        print(Fore.YELLOW + message)
        output.loc[len(output.index)] = [self.Test_DLL_2.__name__, message, 'info_title']
        status = 0
        susdllfname = self.dfdlllist[(self.dfdlllist["Name"].str.match(r"^[a-z0-9]{35,}\.(dll|sys)", case=False) == True)][
            ['PID', 'Process', 'Base', 'Name', 'Path', 'LoadTime']]
        if (susdllfname.empty):
            message = "No DLL(s) or Sys files with unusually long file names were detected"
            print(Fore.GREEN + message)
            output.loc[len(output.index)] = [self.Test_DLL_2.__name__, message, 'info_good']
        else:
            message = "DLL(s) or Sys files with unusually long file names were detected!"
            print(Fore.RED + message)
            output.loc[len(output.index)] = [self.Test_DLL_2.__name__, message, 'info_bad']
            message = susdllfname.to_json(orient='records')
            print(message)
            output.loc[len(output.index)] = [self.Test_DLL_2.__name__, message, 'info_detect']
            status += 1

        return status, output
    def Test_Malw_1(self):
        output = pd.DataFrame(columns=['script', 'message', 'code'])
        message = "MZ header detection within malfind preview"
        print(Fore.YELLOW + message)
        output.loc[len(output.index)] = [self.Test_Malw_1.__name__, message, 'info_title']
        status = 0
        procinject1 = self.dfmalfind[self.dfmalfind["Hexdump"].str.contains("MZ") == True][
            ['PID', 'Process', 'Start VPN', 'End VPN', 'Protection', 'Hexdump', 'Disasm']]
        if (procinject1.empty):
            message = "No MZ header detected in malfind preview output"
            print(Fore.GREEN + message)
            output.loc[len(output.index)] = [self.Test_Malw_1.__name__, message, 'info_good']
        else:
            message = "MZ header detected within malfind preview (Process Injection indicator)"
            print(Fore.RED + message)
            output.loc[len(output.index)] = [self.Test_Malw_1.__name__, message, 'info_bad']
            message = procinject1.to_json(orient='records')
            print(message)
            output.loc[len(output.index)] = [self.Test_Malw_1.__name__, message, 'info_detect']
            status += 1

        return status, output
    def Test_Malw_2(self):
        output = pd.DataFrame(columns=['script', 'message', 'code'])
        message = "Assembly code detection within malfind preview"
        print(Fore.YELLOW + message)
        output.loc[len(output.index)] = [self.Test_Malw_2.__name__, message, 'info_title']
        status = 0
        procinject2 = self.dfmalfind[self.dfmalfind["Hexdump"].str.match(r".*(mov\s+epb,\s?esp).*") == True][
            ['PID', 'Process', 'Start VPN', 'End VPN', 'Protection', 'Hexdump', 'Disasm']]
        if (procinject2.empty):
            message = "No suspicious assembly code detected in malfind preview output"
            print(Fore.GREEN + message)
            output.loc[len(output.index)] = [self.Test_Malw_2.__name__, message, 'info_good']
        else:
            message = "Suspicious assembly code detected in malfind preview output!"
            print(Fore.RED + message)
            output.loc[len(output.index)] = [self.Test_Malw_2.__name__, message, 'info_bad']
            message = procinject2.to_json(orient='records')
            print(message)
            output.loc[len(output.index)] = [self.Test_Malw_2.__name__, message, 'info_detect']
            status += 1

        return status, output

    def Run_All_Tests(self):
        all_func = [self.Test_Proc_1,
                    self.Test_Proc_2,
                    self.Test_Proc_3,
                    self.Test_Proc_4,
                    self.Test_Proc_5,
                    self.Test_Proc_6,
                    self.Test_Proc_7,
                    self.Test_Proc_8,
                    self.Test_Proc_9,
                    self.Test_Net_1,
                    self.Test_Net_2,
                    self.Test_Net_3,
                    self.Test_Net_4,
                    self.Test_Net_5,
                    self.Test_SysT_1,
                    self.Test_SysT_2,
                    self.Test_SysT_3,
                    self.Test_SVC_1,
                    self.Test_SVC_2,
                    self.Test_NamedPipe_1,
                    self.Test_DLL_1,
                    self.Test_DLL_2,
                    self.Test_Malw_1,
                    self.Test_Malw_2
                    ]
        rating = 0
        df = pd.DataFrame(columns=['script', 'message', 'code'])
        df.loc[len(df.index)] = [self.Run_All_Tests.__name__, '', '']
        for func in all_func:
            rate, out = func()
            rating += rate
            df = pd.concat([df, out], ignore_index=True)
            df.reset_index()
        message = "Summary tests rating = " + str(rating)
        print(Fore.CYAN + message)
        df.loc[0] = [self.Run_All_Tests.__name__, message, 'info_info']
        return df