From 281a237e03296ff38c7d0353157fa068c32a8c39 Mon Sep 17 00:00:00 2001 From: Elad Levi <99.elad.levi@gmail.com> Date: Mon, 28 Apr 2025 16:03:45 +0000 Subject: [PATCH] black and ruff fixes --- .../framework/plugins/windows/etwpatch.py | 63 ++++++++++--------- 1 file changed, 33 insertions(+), 30 deletions(-) diff --git a/volatility3/framework/plugins/windows/etwpatch.py b/volatility3/framework/plugins/windows/etwpatch.py index 3605dfccb..e79735213 100644 --- a/volatility3/framework/plugins/windows/etwpatch.py +++ b/volatility3/framework/plugins/windows/etwpatch.py @@ -7,16 +7,17 @@ from volatility3.framework import exceptions, interfaces, renderers from volatility3.framework.configuration import requirements from volatility3.framework.objects import utility from volatility3.framework.renderers import format_hints -from volatility3.plugins.windows import pslist, pe_symbols +from volatility3.plugins.windows import pslist, pe_symbols vollog = logging.getLogger(__name__) + class EtwPatch(interfaces.plugins.PluginInterface): """Identifies ETW (Event Tracing for Windows) patching techniques used by malware to evade detection. - - This plugin examines the first opcode of key ETW functions in ntdll.dll and advapi32.dll - to detect common ETW bypass techniques such as return pointer manipulation (RET) or function - redirection (JMP). Attackers often patch these functions to prevent security tools from + + This plugin examines the first opcode of key ETW functions in ntdll.dll and advapi32.dll + to detect common ETW bypass techniques such as return pointer manipulation (RET) or function + redirection (JMP). Attackers often patch these functions to prevent security tools from receiving telemetry about process execution, API calls, and other system events. """ @@ -28,13 +29,11 @@ class EtwPatch(interfaces.plugins.PluginInterface): pe_symbols.wanted_names_identifier: [ "EtwEventWrite", "EtwEventWriteFull", - "NtTraceEvent" + "NtTraceEvent", ], }, "advapi32.dll": { - pe_symbols.wanted_names_identifier:[ - "EventWrite" - ], + pe_symbols.wanted_names_identifier: ["EventWrite"], }, } @@ -42,22 +41,22 @@ class EtwPatch(interfaces.plugins.PluginInterface): def get_requirements(cls): return [ requirements.ModuleRequirement( - name='kernel', - description='Windows kernel', - architectures=["Intel32", "Intel64"] + name="kernel", + description="Windows kernel", + architectures=["Intel32", "Intel64"], ), requirements.VersionRequirement( name="pslist", component=pslist.PsList, version=(3, 0, 0) ), requirements.VersionRequirement( - name="pe_symbols", component=pslist.PsList, version=(3, 0, 0) + name="pe_symbols", component=pe_symbols.PESymbols, version=(3, 0, 0) ), requirements.ListRequirement( - name='pid', - description='Filter on specific process IDs', + name="pid", + description="Filter on specific process IDs", element_type=int, - optional=True - ) + optional=True, + ), ] def _generator(self): @@ -68,15 +67,15 @@ class EtwPatch(interfaces.plugins.PluginInterface): kernel_module_name=self.config["kernel"], symbols=self.etw_functions, ) - + filter_func = pslist.PsList.create_pid_filter(self.config.get("pid", None)) for proc in pslist.PsList.list_processes( - context=self.context, - kernel_module_name=self.config['kernel'], - filter_func=filter_func, - ): - + context=self.context, + kernel_module_name=self.config["kernel"], + filter_func=filter_func, + ): + try: proc_id = proc.UniqueProcessId proc_name = utility.array_to_string(proc.ImageFileName) @@ -87,16 +86,18 @@ class EtwPatch(interfaces.plugins.PluginInterface): # Map of opcodes to their instruction names opcode_map = { - 'c3': 'RET', - 'e9': 'JMP', + "c3": "RET", + "e9": "JMP", } for dll_name, functions in found_symbols.items(): for func_name, func_addr in functions: try: - opcode = self.context.layers[proc_layer_name].read( - func_addr, 1 - ).hex() + opcode = ( + self.context.layers[proc_layer_name] + .read(func_addr, 1) + .hex() + ) if opcode in opcode_map: instruction = opcode_map[opcode] yield ( @@ -107,11 +108,13 @@ class EtwPatch(interfaces.plugins.PluginInterface): dll_name, func_name, format_hints.Hex(func_addr), - f"{opcode} ({instruction})" + f"{opcode} ({instruction})", ), ) except exceptions.InvalidAddressException: - vollog.debug(f"Invalid address when reading function {func_name} at {func_addr:#x} in process {proc_id}") + vollog.debug( + f"Invalid address when reading function {func_name} at {func_addr:#x} in process {proc_id}" + ) def run(self): return renderers.TreeGrid(