Dumpy.py - master - KuzminVD/Dumpy

Dumpy.py
119 строк4 KB
vladimir
Load code
03 июн 2024, 00:08
03 июн 2024, 00:08f404d0f
from volatility3 import plugins
import volatility3.framework
from volatility3.framework import contexts, interfaces
from volatility3.framework import automagic
from volatility3.framework.plugins import construct_plugin
import logging, os
import pandas as pd

from Analyzer import Analyzer
import TreeGridParser as tgp
from SQLWorker import SQLWorker

import syslog
syslog.openlog(logoption=syslog.LOG_PID, facility=syslog.LOG_WARNING)

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
volatility3.framework.require_interface_version(2, 5, 0)

curr_dir = './'
dump_dir = curr_dir + 'Data examples/1. Dump/'
save_dir = curr_dir + 'Output/'
save_dir_vol = save_dir + 'Volatility Plugins/'
filename = 'flounder-pc-memdump.elf'

def build_context(image_name, context, base_config_path, plugin):
   sw.write_log_to_db("Start [build_context]")
   available_automagics = automagic.available(context)
   log = f"Available automagics : {available_automagics}"
   logger.info(log)
   sw.write_log_to_db(log)
   plugin_config_path = interfaces.configuration.path_join(base_config_path, plugin.__name__)
   automagics = automagic.choose_automagic(available_automagics, plugin)
   context.config['automagic.LayerStacker.stackers'] = automagic.stacker.choose_os_stackers(plugin)
   context.config['automagic.LayerStacker.single_location'] = "file://" + os.getcwd() + "/" + image_name
   constructed = construct_plugin(context, automagics, plugin, base_config_path, None, None)
   log = f"Contructed plugin : {constructed}"
   logger.info(log)
   sw.write_log_to_db(log)
   sw.write_log_to_db("Finish [build_context]")
   return constructed

def run_plugins_and_save(saveDir):
    sw.write_log_to_db("Start [run_plugins_and_save]")
    failures = volatility3.framework.import_files(plugins, True) #Load the framework plugins
    if failures:
        log = f"Some volatility3 plugin couldn't be loaded : {failures}"
    else:
        log = f"Plugins are loaded without failure"
    logger.info(log)
    sw.write_log_to_db(log)

    plugin_list = volatility3.framework.list_plugins()
    plugs = ['windows.pslist.PsList',
             'windows.pstree.PsTree',
             'windows.psscan.PsScan',
             'windows.cmdline.CmdLine',
             'windows.netscan.NetScan',
             'windows.netstat.NetStat',
             'windows.registry.hivescan.HiveScan',
             'windows.registry.hivelist.HiveList',
             'windows.registry.printkey.PrintKey',
             'windows.filescan.FileScan',
             'windows.malfind.Malfind',
             'windows.callbacks.Callbacks',
             'windows.driverscan.DriverScan',
             'windows.dlllist.DllList',
             'windows.svcscan.SvcScan',
             'windows.handles.Handles',
             'windows.ssdt.SSDT',
             'windows.ldrmodules.LdrModules']

    for plug in plugs:
       ctx = contexts.Context()  # Construct a blank context
       base_config_path = "plugins"
       constructed = build_context(dump_dir + filename, ctx, base_config_path, plugin_list[plug])
       log = f"Context configuration {ctx.config}"
       logger.info(log)
       sw.write_log_to_db(log)

       if constructed:
          headers, result = tgp.ArrayParser().parse(constructed.run())
          df = pd.DataFrame(data=result, columns=headers)
          df.to_csv(saveDir + plug + '.csv', index=False, sep=',')

    sw.write_log_to_db("Finish [run_plugins_and_save]")


# init new db if old exists
sw = SQLWorker(save_dir_vol, save_dir, 'temp.db')
sw.init_db()

# run plugins to extract data from dump to csv files
run_plugins_and_save(save_dir_vol)

# write csv files data to db tables
sw.write_log_to_db("Start [write_csv_to_db]")
sw.write_csv_to_db('windows.*.*.csv')
sw.write_log_to_db("Finish [write_csv_to_db]")

# get the results of all tests
sw.write_log_to_db("Start [Run_All_Tests]")
an = Analyzer(save_dir_vol, save_dir)
results = an.Run_All_Tests()
sw.write_log_to_db("Finish [Run_All_Tests]")

# save results to csv file
sw.write_log_to_db("Saving results to csv: start")
results.to_csv(save_dir + 'All_Tests' + '.csv', index=False, sep=";")
sw.write_log_to_db("Saving results to csv: finish")

# write tests results to db
sw.write_log_to_db("Start [write_results_to_db]")
sw.write_results_to_db(results)
sw.write_log_to_db("Finish [write_results_to_db]")

# write logs to syslog
for index, row in results.iterrows():
    syslog.syslog(' | '.join(row))