From fdb131f870f5a3ef97dfe5f33e3ee05cf17d83f2 Mon Sep 17 00:00:00 2001 From: Dave Lassalle Date: Thu, 1 Aug 2024 16:27:48 -0500 Subject: [PATCH] #816 - add cmdscan and bug fixes --- .../framework/plugins/windows/cmdscan.py | 327 ++++++++++++++++++ .../framework/plugins/windows/consoles.py | 2 +- .../consoles-win10-20348-2461-x64.json | 2 +- .../consoles/consoles-win10-20348-x64.json | 2 +- .../symbols/windows/extensions/consoles.py | 31 +- 5 files changed, 360 insertions(+), 4 deletions(-) create mode 100644 volatility3/framework/plugins/windows/cmdscan.py diff --git a/volatility3/framework/plugins/windows/cmdscan.py b/volatility3/framework/plugins/windows/cmdscan.py new file mode 100644 index 000000000..86a34a4f8 --- /dev/null +++ b/volatility3/framework/plugins/windows/cmdscan.py @@ -0,0 +1,327 @@ +# This file is Copyright 2024 Volatility Foundation and licensed under the Volatility Software License 1.0 +# which is available at https://www.volatilityfoundation.org/license/vsl-v1.0 +# + +# This module attempts to locate windows console histories. + +import logging +import struct +from typing import Tuple, Generator, Set, Dict, Any, List, Optional + +from volatility3.framework import interfaces +from volatility3.framework import renderers +from volatility3.framework.configuration import requirements +from volatility3.framework.layers import scanners +from volatility3.framework.objects import utility +from volatility3.framework.renderers import format_hints +from volatility3.plugins.windows import pslist, vadinfo, info, verinfo, consoles +from volatility3.plugins.windows.registry import hivelist + + +try: + import capstone + + has_capstone = True +except ImportError: + has_capstone = False + +vollog = logging.getLogger(__name__) + + +class CmdScan(interfaces.plugins.PluginInterface): + """Looks for Windows Command History lists""" + + _required_framework_version = (2, 4, 0) + + @classmethod + def get_requirements(cls): + # Since we're calling the plugin, make sure we have the plugin's requirements + return [ + requirements.ModuleRequirement( + name="kernel", + description="Windows kernel", + architectures=["Intel32", "Intel64"], + ), + requirements.VersionRequirement( + name="pslist", component=pslist.PsList, version=(2, 0, 0) + ), + requirements.PluginRequirement( + name="hivelist", plugin=hivelist.HiveList, version=(1, 0, 0) + ), + requirements.BooleanRequirement( + name="no_registry", + description="Don't search the registry for possible values of CommandHistorySize", + optional=True, + default=False, + ), + requirements.ListRequirement( + name="max_history", + element_type=int, + description="CommandHistorySize values to search for.", + optional=True, + default=[50], + ), + ] + + @classmethod + def get_filtered_vads( + cls, conhost_proc: interfaces.context.ContextInterface, size_filter: Optional[int]=0x40000000 + ) -> List[Tuple[int, int]]: + """ + Returns vads of a process with smaller than size_filter + + Args: + conhost_proc: the process object for conhost.exe + + Returns: + A list of tuples of: + vad_base: the base address + vad_size: the size of the VAD + """ + vads = [] + for vad in conhost_proc.get_vad_root().traverse(): + base = vad.get_start() + if vad.get_size() < size_filter: + vads.append((base, vad.get_size())) + + return vads + + @classmethod + def get_command_history( + cls, + context: interfaces.context.ContextInterface, + kernel_layer_name: str, + kernel_table_name: str, + config_path: str, + procs: Generator[interfaces.objects.ObjectInterface, None, None], + max_history: Set[int], + ) -> Tuple[ + interfaces.context.ContextInterface, + interfaces.context.ContextInterface, + Dict[str, Any], + ]: + """Gets the list of commands from each Command History structure + + Args: + context: The context to retrieve required elements (layers, symbol tables) from + kernel_layer_name: The name of the layer on which to operate + kernel_table_name: The name of the table containing the kernel symbols + config_path: The config path where to find symbol files + procs: list of process objects + max_history: an initial set of CommandHistorySize values + + Returns: + The conhost process object, the command history structure, a dictionary of properties for + that command historyn structure. + """ + + conhost_symbol_table = consoles.Consoles.create_conhost_symbol_table( + context, kernel_layer_name, kernel_table_name, config_path + ) + + for conhost_proc, proc_layer_name in consoles.Consoles.find_conhost_proc(procs): + if not conhost_proc: + vollog.info( + "Unable to find a valid conhost.exe process in the process list. Analysis cannot proceed." + ) + continue + vollog.debug( + f"Found conhost process {conhost_proc} with pid {conhost_proc.UniqueProcessId}" + ) + + conhostexe_base, conhostexe_size = consoles.Consoles.find_conhostexe(conhost_proc) + if not conhostexe_base: + vollog.info( + "Unable to find the location of conhost.exe. Analysis cannot proceed." + ) + continue + vollog.debug(f"Found conhost.exe base at {conhostexe_base:#x}") + + proc_layer = context.layers[proc_layer_name] + + conhost_module = context.module( + conhost_symbol_table, proc_layer_name, offset=conhostexe_base + ) + + sections = cls.get_filtered_vads(conhost_proc) + # scan for potential _COMMAND_HISTORY structures by using the CommandHistorySize + for max_history_value in max_history: + max_history_bytes = struct.pack("H", max_history_value) + vollog.debug( + f"Scanning for CommandHistorySize value: {max_history_bytes}" + ) + for address in proc_layer.scan( + context, + scanners.BytesScanner(max_history_bytes), + sections=sections, + ): + command_history_properties = [] + + try: + command_history = conhost_module.object( + "_COMMAND_HISTORY", + offset=address + - conhost_module.get_type( + "_COMMAND_HISTORY" + ).relative_child_offset("CommandCountMax"), + absolute=True, + ) + + if not command_history.is_valid(max_history_value): + continue + + vollog.debug( + f"Getting Command History properties for {command_history}" + ) + command_history_properties.append( + { + "name": f"_COMMAND_HISTORY.Application", + "address": command_history.Application.vol.offset, + "data": command_history.get_application(), + } + ) + command_history_properties.append( + { + "name": f"_COMMAND_HISTORY.ProcessHandle", + "address": command_history.ConsoleProcessHandle.ProcessHandle.vol.offset, + "data": hex( + command_history.ConsoleProcessHandle.ProcessHandle + ), + } + ) + command_history_properties.append( + { + "name": f"_COMMAND_HISTORY.CommandCount", + "address": None, + "data": command_history.CommandCount, + } + ) + command_history_properties.append( + { + "name": f"_COMMAND_HISTORY.LastDisplayed", + "address": command_history.LastDisplayed.vol.offset, + "data": command_history.LastDisplayed, + } + ) + command_history_properties.append( + { + "name": f"_COMMAND_HISTORY.CommandCountMax", + "address": command_history.CommandCountMax.vol.offset, + "data": command_history.CommandCountMax, + } + ) + + command_history_properties.append( + { + "name": f"_COMMAND_HISTORY.CommandBucket", + "address": command_history.CommandBucket.vol.offset, + "data": "", + } + ) + for ( + cmd_index, + bucket_cmd, + ) in command_history.scan_command_bucket(): + try: + command_history_properties.append( + { + "name": f"_COMMAND_HISTORY.CommandBucket_Command_{cmd_index}", + "address": bucket_cmd.vol.offset, + "data": bucket_cmd.get_command(), + } + ) + except Exception as e: + vollog.debug( + f"reading {bucket_cmd} encountered exception {e}" + ) + except Exception as e: + vollog.debug( + f"reading {command_history} encountered exception {e}" + ) + + yield conhost_proc, command_history, command_history_properties + + def _generator( + self, procs: Generator[interfaces.objects.ObjectInterface, None, None] + ): + """ + Generates the command history to use in rendering + + Args: + procs: the process list filtered to conhost.exe instances + """ + + kernel = self.context.modules[self.config["kernel"]] + + max_history = set(self.config.get("max_history", [50])) + no_registry = self.config.get("no_registry") + + if no_registry is False: + max_history, _max_buffers = consoles.Consoles.get_console_settings_from_registry( + self.context, + self.config_path, + kernel.layer_name, + kernel.symbol_table_name, + max_history, + [], + ) + + vollog.debug(f"Possible CommandHistorySize values: {max_history}") + + for proc, command_history, command_history_properties in self.get_command_history( + self.context, + kernel.layer_name, + kernel.symbol_table_name, + self.config_path, + procs, + max_history, + ): + process_name = utility.array_to_string(proc.ImageFileName) + + if command_history and command_history_properties: + for command_history_property in command_history_properties: + yield ( + 0, + ( + proc.UniqueProcessId, + process_name, + format_hints.Hex(command_history.vol.offset), + command_history_property["name"], + ( + renderers.NotApplicableValue() + if command_history_property["address"] is None + else format_hints.Hex(command_history_property["address"]) + ), + str(command_history_property["data"]), + ), + ) + + def _conhost_proc_filter(self, proc): + """ + Used to filter to only conhost.exe processes + """ + process_name = utility.array_to_string(proc.ImageFileName) + + return process_name != "conhost.exe" + + def run(self): + kernel = self.context.modules[self.config["kernel"]] + + return renderers.TreeGrid( + [ + ("PID", int), + ("Process", str), + ("ConsoleInfo", format_hints.Hex), + ("Property", str), + ("Address", format_hints.Hex), + ("Data", str), + ], + self._generator( + pslist.PsList.list_processes( + context=self.context, + layer_name=kernel.layer_name, + symbol_table=kernel.symbol_table_name, + filter_func=self._conhost_proc_filter, + ) + ), + ) diff --git a/volatility3/framework/plugins/windows/consoles.py b/volatility3/framework/plugins/windows/consoles.py index f6bbb3ae7..509b6ae9d 100644 --- a/volatility3/framework/plugins/windows/consoles.py +++ b/volatility3/framework/plugins/windows/consoles.py @@ -363,7 +363,7 @@ class Consoles(interfaces.plugins.PluginInterface): interfaces.context.ContextInterface, Dict[str, Any], ]: - """Extracts the cmdline from PEB + """Gets the Console Information structure and its related properties for each conhost process Args: context: The context to retrieve required elements (layers, symbol tables) from diff --git a/volatility3/framework/symbols/windows/consoles/consoles-win10-20348-2461-x64.json b/volatility3/framework/symbols/windows/consoles/consoles-win10-20348-2461-x64.json index 3ae41fdd7..a4adf8028 100644 --- a/volatility3/framework/symbols/windows/consoles/consoles-win10-20348-2461-x64.json +++ b/volatility3/framework/symbols/windows/consoles/consoles-win10-20348-2461-x64.json @@ -232,7 +232,7 @@ "kind": "base", "name": "unsigned int" }, - "offset": 20 + "offset": 24 } }, "kind": "struct", diff --git a/volatility3/framework/symbols/windows/consoles/consoles-win10-20348-x64.json b/volatility3/framework/symbols/windows/consoles/consoles-win10-20348-x64.json index b74862598..5cc7ab6f5 100644 --- a/volatility3/framework/symbols/windows/consoles/consoles-win10-20348-x64.json +++ b/volatility3/framework/symbols/windows/consoles/consoles-win10-20348-x64.json @@ -232,7 +232,7 @@ "kind": "base", "name": "unsigned int" }, - "offset": 20 + "offset": 24 } }, "kind": "struct", diff --git a/volatility3/framework/symbols/windows/extensions/consoles.py b/volatility3/framework/symbols/windows/extensions/consoles.py index 2db1f4cad..279ee723e 100644 --- a/volatility3/framework/symbols/windows/extensions/consoles.py +++ b/volatility3/framework/symbols/windows/extensions/consoles.py @@ -220,6 +220,12 @@ class CONSOLE_INFORMATION(objects.StructType): class COMMAND(objects.StructType): """A Command Structure""" + def is_valid(self): + if self.Length < 1 or self.Allocated < 1 or self.Length > 1024 or self.Allocated > 1024: + return False + + return True + def get_command(self): if self.Length < 8: return self.Chars.cast( @@ -243,6 +249,26 @@ class COMMAND_HISTORY(objects.StructType): command_size = self._context.symbol_space.get_type(command_type).size return int((self.CommandBucket.End - self.CommandBucket.Begin) / command_size) + @property + def ProcessHandle(self): + """ Allow ProcessHandle to be referenced regardless of OS version """ + return self.ConsoleProcessHandle.ProcessHandle + + def is_valid(self, max_history=50): + # The count must be between zero and max + if self.CommandCount < 0 or self.CommandCount > max_history: + return False + + # Last displayed must be between -1 and max + if self.LastDisplayed < -1 or self.LastDisplayed > max_history: + return False + + # Process handle must be a valid pid + if self.ProcessHandle <= 0 or self.ProcessHandle > 0xFFFF or self.ProcessHandle % 4 != 0: + return False + + return True + def get_application(self): if self.Application.Length < 8: return self.Application.Chars.cast( @@ -265,6 +291,7 @@ class COMMAND_HISTORY(objects.StructType): self.vol.type_name ).size command_size = self._context.symbol_space.get_type(command_type).size + if end is None: end = max( self.CommandBucket.EndCapacity, @@ -272,7 +299,9 @@ class COMMAND_HISTORY(objects.StructType): ) for i, pointer in enumerate(range(self.CommandBucket.Begin, end, command_size)): - yield i, self._context.object(command_type, self.vol.layer_name, pointer) + cmd = self._context.object(command_type, self.vol.layer_name, pointer) + if cmd.is_valid(): + yield i, cmd def get_commands(self): """Generator for commands in the history buffer.