From a3caf39097984a8180c08d5dd50d49208e2ec917 Mon Sep 17 00:00:00 2001 From: Mike Auty Date: Thu, 29 Aug 2019 01:08:16 +0100 Subject: [PATCH] Use a mapping to determine kind to class. --- doc/source/complex-plugin.rst | 2 +- volatility/framework/interfaces/symbols.py | 2 +- volatility/framework/layers/registry.py | 9 ++-- volatility/framework/objects/__init__.py | 43 +++++++++++++------ .../plugins/windows/registry/printkey.py | 2 +- .../framework/symbols/generic/__init__.py | 2 +- volatility/framework/symbols/intermed.py | 14 ++++-- .../symbols/linux/extensions/__init__.py | 22 +++++----- .../symbols/linux/extensions/bash.py | 2 +- .../symbols/mac/extensions/__init__.py | 24 +++-------- .../symbols/windows/extensions/__init__.py | 26 +++++------ .../symbols/windows/extensions/kdbg.py | 2 +- .../symbols/windows/extensions/pe.py | 4 +- .../symbols/windows/extensions/registry.py | 19 ++++---- .../symbols/windows/extensions/services.py | 36 ++++++++++------ 15 files changed, 117 insertions(+), 92 deletions(-) diff --git a/doc/source/complex-plugin.rst b/doc/source/complex-plugin.rst index 38feead2a..6f34d3488 100644 --- a/doc/source/complex-plugin.rst +++ b/doc/source/complex-plugin.rst @@ -112,7 +112,7 @@ Another useful parameter is `table_mapping` which allows for type referenced ins table_mapping = {'one_table': 'another_table'}) The last parameter that can be used is called `class_types` which allows a particular structure to be instantiated on -a class other than :py:class:`~volatility.framework.objects.Struct`, allowing for additional methods to be defined and +a class other than :py:class:`~volatility.framework.objects.StructType`, allowing for additional methods to be defined and associated with the type. The table name can then by used to access the constructed table from the context, such as: diff --git a/volatility/framework/interfaces/symbols.py b/volatility/framework/interfaces/symbols.py index bc8a1c506..1b6bfa7b1 100644 --- a/volatility/framework/interfaces/symbols.py +++ b/volatility/framework/interfaces/symbols.py @@ -249,7 +249,7 @@ class SymbolTableInterface(BaseSymbolTableInterface, configuration.ConfigurableI isf_url: The URL pointing to the ISF file location native_types: The NativeSymbolTable that contains the native types for this symbol table table_mapping: A dictionary linking names referenced in the file with symbol tables in the context - class_types: A dictionary of type names and classes that override Struct when they are instantiated + class_types: A dictionary of type names and classes that override StructType when they are instantiated """ configuration.ConfigurableInterface.__init__(self, context, config_path) BaseSymbolTableInterface.__init__(self, name, native_types, table_mapping, class_types = class_types) diff --git a/volatility/framework/layers/registry.py b/volatility/framework/layers/registry.py index e1017a7fc..005f50dda 100644 --- a/volatility/framework/layers/registry.py +++ b/volatility/framework/layers/registry.py @@ -98,7 +98,7 @@ class RegistryHive(linear.LinearlyMappedLayer): pass return 0x20 - def get_cell(self, cell_offset: int) -> 'objects.Struct': + def get_cell(self, cell_offset: int) -> 'objects.StructType': """Returns the appropriate Cell value for a cell offset""" # This would be an _HCELL containing CELL_DATA, but to save time we skip the size of the HCELL cell = self._context.object( @@ -107,7 +107,7 @@ class RegistryHive(linear.LinearlyMappedLayer): layer_name = self.name) return cell - def get_node(self, cell_offset: int) -> 'objects.Struct': + def get_node(self, cell_offset: int) -> 'objects.StructType': """Returns the appropriate Node, interpreted from the Cell based on its Signature""" cell = self.get_cell(cell_offset) signature = cell.cast('string', max_length = 2, encoding = 'latin-1') @@ -129,7 +129,7 @@ class RegistryHive(linear.LinearlyMappedLayer): cell_offset)) return cell - def get_key(self, key: str, return_list: bool = False) -> Union[List[objects.Struct], objects.Struct]: + def get_key(self, key: str, return_list: bool = False) -> Union[List[objects.StructType], objects.StructType]: """Gets a specific registry key by key path return_list specifies whether the return result will be a single node (default) or a list of nodes from @@ -157,7 +157,8 @@ class RegistryHive(linear.LinearlyMappedLayer): return node_key return node_key[-1] - def visit_nodes(self, visitor: Callable[[objects.Struct], None], node: Optional[objects.Struct] = None) -> None: + def visit_nodes(self, visitor: Callable[[objects.StructType], None], + node: Optional[objects.StructType] = None) -> None: """Applies a callable (visitor) to all nodes within the registry tree from a given node""" if not node: node = self.get_node(self.root_cell_offset) diff --git a/volatility/framework/objects/__init__.py b/volatility/framework/objects/__init__.py index edd22a433..25961ff1c 100644 --- a/volatility/framework/objects/__init__.py +++ b/volatility/framework/objects/__init__.py @@ -558,7 +558,7 @@ class Array(interfaces.objects.ObjectInterface, abc.Sequence): raise NotImplementedError("Writing to Arrays is not yet implemented") -class Struct(interfaces.objects.ObjectInterface): +class AggregateType(interfaces.objects.ObjectInterface): """Object which can contain members that are other objects Keep the number of methods in this class low or very specific, since each one could overload a valid member. @@ -582,7 +582,7 @@ class Struct(interfaces.objects.ObjectInterface): def size(cls, template: interfaces.objects.Template) -> int: """Method to return the size of this type""" if template.vol.get('size', None) is None: - raise TypeError("Struct ObjectTemplate not provided with a size") + raise TypeError("ObjectTemplate not provided with a size") return template.vol.size @classmethod @@ -623,10 +623,17 @@ class Struct(interfaces.objects.ObjectInterface): def _check_members(cls, members: Dict[str, Tuple[int, interfaces.objects.Template]]) -> None: # Members should be an iterable mapping of symbol names to tuples of (relative_offset, ObjectTemplate) # An object template is a callable that when called with a context, offset, layer_name and type_name + + # We duplicate this code to avoid polluting the methodspace + agg_name = 'AggregateType' + for agg_type in AggregateTypes: + if isinstance(cls, agg_type): + agg_name = agg_type.__name__ + assert isinstance(members, abc.Mapping) - "Struct members parameter must be a mapping: {}".format(type(members)) + "{} members parameter must be a mapping: {}".format(agg_name, type(members)) assert all([(isinstance(member, tuple) and len(member) == 2) for member in members.values()]) - "Struct members must be a tuple of relative_offsets and templates" + "{} members must be a tuple of relative_offsets and templates".format(agg_name) def member(self, attr: str = 'member') -> object: """Specifically named method for retrieving members.""" @@ -649,25 +656,37 @@ class Struct(interfaces.objects.ObjectInterface): native_layer_name = self.vol.native_layer_name)) self._concrete_members[attr] = member return member - raise AttributeError("Struct has no attribute: {}.{}".format(self.vol.type_name, attr)) + # We duplicate this code to avoid polluting the methodspace + agg_name = 'AggregateType' + for agg_type in AggregateTypes: + if isinstance(self, agg_type): + agg_name = agg_type.__name__ + raise AttributeError("{} has no attribute: {}.{}".format(agg_name, self.vol.type_name, attr)) def __dir__(self) -> Iterable[str]: """Returns a complete list of members when dir is called""" return list(super().__dir__()) + list(self.vol.members) def write(self, value): - raise TypeError("Structs cannot be written to directly, individual members must be written instead") + # We duplicate this code to avoid polluting the methodspace + agg_name = 'AggregateType' + for agg_type in AggregateTypes: + if isinstance(self, agg_type): + agg_name = agg_type.__name__ + raise TypeError( + "{}s cannot be written to directly, individual members must be written instead".format(agg_name)) -# Nice way of duplicating the class, but *could* causes problems with isintance -class Union(Struct): +class StructType(AggregateType): pass -class CPPObject(Struct): +class UnionType(AggregateType): pass -# Really nasty way of duplicating the class -# WILL cause problems with any mutable class/static variables -# Union = type('Union', Struct.__bases__, dict(Struct.__dict__)) +class ClassType(AggregateType): + pass + + +AggregateTypes = {StructType: 'struct', UnionType: 'union', ClassType: 'class'} diff --git a/volatility/framework/plugins/windows/registry/printkey.py b/volatility/framework/plugins/windows/registry/printkey.py index 2e317a9aa..6b151260e 100644 --- a/volatility/framework/plugins/windows/registry/printkey.py +++ b/volatility/framework/plugins/windows/registry/printkey.py @@ -44,7 +44,7 @@ class PrintKey(interfaces.plugins.PluginInterface): return key_path.count("\\", key_arg_ending_index) - def hive_walker(self, hive: RegistryHive, node_path: Sequence[objects.Struct] = None, + def hive_walker(self, hive: RegistryHive, node_path: Sequence[objects.StructType] = None, key_path: str = None) -> Generator: """Walks through a set of nodes from a given node (last one in node_path). Avoids loops by not traversing into nodes already present in the node_path diff --git a/volatility/framework/symbols/generic/__init__.py b/volatility/framework/symbols/generic/__init__.py index b290b4115..01e7f709d 100644 --- a/volatility/framework/symbols/generic/__init__.py +++ b/volatility/framework/symbols/generic/__init__.py @@ -9,7 +9,7 @@ from typing import Union from volatility.framework import objects, interfaces -class GenericIntelProcess(objects.Struct): +class GenericIntelProcess(objects.StructType): def _add_process_layer(self, context: interfaces.context.ContextInterface, diff --git a/volatility/framework/symbols/intermed.py b/volatility/framework/symbols/intermed.py index 826f2cf9d..a6982b186 100644 --- a/volatility/framework/symbols/intermed.py +++ b/volatility/framework/symbols/intermed.py @@ -95,7 +95,7 @@ class IntermediateSymbolTable(interfaces.symbols.SymbolTableInterface): native_types: The NativeSymbolTable that contains the native types for this symbol table table_mapping: A dictionary linking names referenced in the file with symbol tables in the context validate: Determines whether the ISF file will be validated against the appropriate schema - class_types: A dictionary of type names and classes that override Struct when they are instantiated + class_types: A dictionary of type names and classes that override StructType when they are instantiated """ # Check there are no obvious errors # Open the file and test the version @@ -327,7 +327,7 @@ class Version1Format(ISFormatTable): return list(self._json_object.get('user_types', {})) + list(self.natives.types) def get_type_class(self, name: str) -> Type[interfaces.objects.ObjectInterface]: - return self._overrides.get(name, objects.Struct) + return self._overrides.get(name, objects.AggregateType) def set_type_class(self, name: str, clazz: Type[interfaces.objects.ObjectInterface]) -> None: if name not in self.types: @@ -372,7 +372,7 @@ class Version1Format(ISFormatTable): return native_template # Otherwise - if dictionary['kind'] not in ['struct', 'union', 'CPPObject']: + if dictionary['kind'] not in objects.AggregateTypes.values(): raise exceptions.SymbolSpaceError("Unknown Intermediate format: {}".format(dictionary)) reference_name = dictionary['name'] @@ -423,6 +423,10 @@ class Version1Format(ISFormatTable): member = (interdict['offset'], self._interdict_to_template(interdict['type'])) members[member_name] = member object_class = self.get_type_class(type_name) + if object_class == objects.AggregateType: + for clazz in objects.AggregateTypes: + if objects.AggregateTypes[clazz] == curdict['kind']: + object_class = clazz return objects.templates.ObjectTemplate( type_name = self.name + constants.BANG + type_name, object_class = object_class, @@ -471,6 +475,10 @@ class Version2Format(Version1Format): member = (interdict['offset'], self._interdict_to_template(interdict['type'])) members[member_name] = member object_class = self.get_type_class(type_name) + if object_class == objects.AggregateType: + for clazz in objects.AggregateTypes: + if objects.AggregateTypes[clazz] == curdict['kind']: + object_class = clazz return objects.templates.ObjectTemplate( type_name = self.name + constants.BANG + type_name, object_class = object_class, diff --git a/volatility/framework/symbols/linux/extensions/__init__.py b/volatility/framework/symbols/linux/extensions/__init__.py index eada8eb90..7640267e4 100644 --- a/volatility/framework/symbols/linux/extensions/__init__.py +++ b/volatility/framework/symbols/linux/extensions/__init__.py @@ -77,7 +77,7 @@ class task_struct(generic.GenericIntelProcess): yield (start, end - start) -class fs_struct(objects.Struct): +class fs_struct(objects.StructType): def get_root_dentry(self): # < 2.6.26 @@ -98,7 +98,7 @@ class fs_struct(objects.Struct): raise AttributeError("Unable to find the root mount") -class mm_struct(objects.Struct): +class mm_struct(objects.StructType): def get_mmap_iter(self) -> Iterable[interfaces.objects.ObjectInterface]: """Returns an iterator for the mmap list member of an mm_struct.""" @@ -117,7 +117,7 @@ class mm_struct(objects.Struct): link = link.vm_next -class super_block(objects.Struct): +class super_block(objects.StructType): # include/linux/kdev_t.h MINORBITS = 20 @@ -130,7 +130,7 @@ class super_block(objects.Struct): return self.s_dev & ((1 << self.MINORBITS) - 1) -class vm_area_struct(objects.Struct): +class vm_area_struct(objects.StructType): perm_flags = { 0x00000001: "r", 0x00000002: "w", @@ -228,7 +228,7 @@ class vm_area_struct(objects.Struct): return ret -class qstr(objects.Struct): +class qstr(objects.StructType): def name_as_str(self) -> str: if self.has_member("len"): @@ -244,13 +244,13 @@ class qstr(objects.Struct): return ret -class dentry(objects.Struct): +class dentry(objects.StructType): def path(self) -> str: return self.d_name.name_as_str() -class struct_file(objects.Struct): +class struct_file(objects.StructType): def get_dentry(self) -> interfaces.objects.ObjectInterface: if self.has_member("f_dentry"): @@ -269,7 +269,7 @@ class struct_file(objects.Struct): raise AttributeError("Unable to find file -> vfs mount") -class list_head(objects.Struct, collections.abc.Iterable): +class list_head(objects.StructType, collections.abc.Iterable): def to_list(self, symbol_type: str, @@ -303,7 +303,7 @@ class list_head(objects.Struct, collections.abc.Iterable): return self.to_list(self.vol.parent.vol.type_name, self.vol.member_name) -class files_struct(objects.Struct): +class files_struct(objects.StructType): def get_fds(self) -> interfaces.objects.ObjectInterface: if self.has_member("fdt"): @@ -322,7 +322,7 @@ class files_struct(objects.Struct): raise AttributeError("Unable to find files -> maximum file descriptors") -class mount(objects.Struct): +class mount(objects.StructType): def get_mnt_sb(self): if self.has_member("mnt"): @@ -355,7 +355,7 @@ class mount(objects.Struct): return self.mnt_mountpoint -class vfsmount(objects.Struct): +class vfsmount(objects.StructType): def is_valid(self): return self.get_mnt_sb() != 0 and \ diff --git a/volatility/framework/symbols/linux/extensions/bash.py b/volatility/framework/symbols/linux/extensions/bash.py index 97d22332f..4a4a84319 100644 --- a/volatility/framework/symbols/linux/extensions/bash.py +++ b/volatility/framework/symbols/linux/extensions/bash.py @@ -8,7 +8,7 @@ from volatility.framework.objects import utility from volatility.framework.renderers import conversion -class hist_entry(objects.Struct): +class hist_entry(objects.StructType): def is_valid(self): try: diff --git a/volatility/framework/symbols/mac/extensions/__init__.py b/volatility/framework/symbols/mac/extensions/__init__.py index 72a8ced14..43014a715 100644 --- a/volatility/framework/symbols/mac/extensions/__init__.py +++ b/volatility/framework/symbols/mac/extensions/__init__.py @@ -77,7 +77,7 @@ class proc(generic.GenericIntelProcess): yield (start, end - start) -class fileglob(objects.Struct): +class fileglob(objects.StructType): def get_fg_type(self): ret = "INVALID" @@ -92,7 +92,7 @@ class fileglob(objects.Struct): return ret.description -class vm_map_object(objects.Struct): +class vm_map_object(objects.StructType): def get_map_object(self): if self.has_member("vm_object"): @@ -103,7 +103,7 @@ class vm_map_object(objects.Struct): raise AttributeError("vm_map_object -> get_object") -class vnode(objects.Struct): +class vnode(objects.StructType): def _do_calc_path(self, ret, vnodeobj, vname): if vnodeobj is None: @@ -138,7 +138,7 @@ class vnode(objects.Struct): return ret.decode("utf-8") -class vm_map_entry(objects.Struct): +class vm_map_entry(objects.StructType): def is_suspicious(self, context, config_prefix): """Flags memory regions that are mapped rwx or that map an executable not back from a file on disk""" @@ -263,7 +263,7 @@ class vm_map_entry(objects.Struct): return ret -class socket(objects.Struct): +class socket(objects.StructType): def get_inpcb(self): try: @@ -320,7 +320,8 @@ class socket(objects.Struct): return ret -class inpcb(objects.Struct): + +class inpcb(objects.StructType): def get_tcp_state(self): tcp_states = ("CLOSED", "LISTEN", "SYN_SENT", "SYN_RECV", "ESTABLISHED", "CLOSE_WAIT", "FIN_WAIT1", "CLOSING", @@ -356,14 +357,3 @@ class inpcb(objects.Struct): rport = self.inp_fport return [lip, lport, rip, rport] - -class queue_entry(objects.Struct): - def walk_list(self, list_head, member_name, type_name): - n = self.next.dereference().cast(type_name) - while n is not None and n.vol.offset != list_head: - yield n - n = n.member(attr = member_name).next.dereference().cast(type_name) - p = self.prev.dereference().cast(type_name) - while p is not None and p.vol.offset != list_head: - yield p - p = p.member(attr = member_name).prev.dereference().cast(type_name) diff --git a/volatility/framework/symbols/windows/extensions/__init__.py b/volatility/framework/symbols/windows/extensions/__init__.py index e495b103e..35011d8cc 100644 --- a/volatility/framework/symbols/windows/extensions/__init__.py +++ b/volatility/framework/symbols/windows/extensions/__init__.py @@ -18,7 +18,7 @@ vollog = logging.getLogger(__name__) # Keep these in a basic module, to prevent import cycles when symbol providers require them -class _POOL_HEADER(objects.Struct): +class _POOL_HEADER(objects.StructType): """A kernel pool allocation header. Exists at the base of the allocation and provides a tag that we can scan for.""" @@ -113,7 +113,7 @@ class _POOL_HEADER(objects.Struct): return None -class _KSYSTEM_TIME(objects.Struct): +class _KSYSTEM_TIME(objects.StructType): """A system time structure that stores a high and low part.""" def get_time(self): @@ -121,7 +121,7 @@ class _KSYSTEM_TIME(objects.Struct): return conversion.wintime_to_datetime(wintime) -class _MMVAD_SHORT(objects.Struct): +class _MMVAD_SHORT(objects.StructType): """A class that represents process virtual memory ranges. Each instance is a node in a binary tree structure and is pointed to by VadRoot.""" @@ -375,7 +375,7 @@ class _MMVAD(_MMVAD_SHORT): return file_name -class _EX_FAST_REF(objects.Struct): +class _EX_FAST_REF(objects.StructType): """This is a standard Windows structure that stores a pointer to an object but also leverages the least significant bits to encode additional details. When dereferencing the pointer, we need to strip off the extra bits.""" @@ -416,7 +416,7 @@ class ExecutiveObject(interfaces.objects.ObjectInterface): native_layer_name = self.vol.native_layer_name) -class _DEVICE_OBJECT(objects.Struct, ExecutiveObject): +class _DEVICE_OBJECT(objects.StructType, ExecutiveObject): """A class for kernel device objects.""" def get_device_name(self) -> str: @@ -424,7 +424,7 @@ class _DEVICE_OBJECT(objects.Struct, ExecutiveObject): return header.NameInfo.Name.String # type: ignore -class _DRIVER_OBJECT(objects.Struct, ExecutiveObject): +class _DRIVER_OBJECT(objects.StructType, ExecutiveObject): """A class for kernel driver objects.""" def get_driver_name(self) -> str: @@ -436,7 +436,7 @@ class _DRIVER_OBJECT(objects.Struct, ExecutiveObject): return True -class _OBJECT_SYMBOLIC_LINK(objects.Struct, ExecutiveObject): +class _OBJECT_SYMBOLIC_LINK(objects.StructType, ExecutiveObject): """A class for kernel link objects.""" def get_link_name(self) -> str: @@ -451,7 +451,7 @@ class _OBJECT_SYMBOLIC_LINK(objects.Struct, ExecutiveObject): return conversion.wintime_to_datetime(self.CreationTime.QuadPart) -class _FILE_OBJECT(objects.Struct, ExecutiveObject): +class _FILE_OBJECT(objects.StructType, ExecutiveObject): """A class for windows file objects""" def is_valid(self) -> bool: @@ -472,7 +472,7 @@ class _FILE_OBJECT(objects.Struct, ExecutiveObject): return name -class _KMUTANT(objects.Struct, ExecutiveObject): +class _KMUTANT(objects.StructType, ExecutiveObject): """A class for windows mutant objects""" def is_valid(self) -> bool: @@ -485,7 +485,7 @@ class _KMUTANT(objects.Struct, ExecutiveObject): return header.NameInfo.Name.String # type: ignore -class _OBJECT_HEADER(objects.Struct): +class _OBJECT_HEADER(objects.StructType): """A class for the headers for executive kernel objects, which contains quota information, ownership details, naming data, and ACLs.""" @@ -557,7 +557,7 @@ class _OBJECT_HEADER(objects.Struct): return header -class _ETHREAD(objects.Struct): +class _ETHREAD(objects.StructType): """A class for executive thread objects.""" def owning_process(self, kernel_layer: str = None) -> interfaces.objects.ObjectInterface: @@ -565,7 +565,7 @@ class _ETHREAD(objects.Struct): return self.ThreadsProcess.dereference(kernel_layer) -class _UNICODE_STRING(objects.Struct): +class _UNICODE_STRING(objects.StructType): """A class for Windows unicode string structures.""" def get_string(self) -> interfaces.objects.ObjectInterface: @@ -733,7 +733,7 @@ class _EPROCESS(generic.GenericIntelProcess, ExecutiveObject): return self.VadRoot.dereference().cast("_MMVAD") -class _LIST_ENTRY(objects.Struct, collections.abc.Iterable): +class _LIST_ENTRY(objects.StructType, collections.abc.Iterable): """A class for double-linked lists on Windows.""" def to_list(self, diff --git a/volatility/framework/symbols/windows/extensions/kdbg.py b/volatility/framework/symbols/windows/extensions/kdbg.py index 6595c732a..653953417 100644 --- a/volatility/framework/symbols/windows/extensions/kdbg.py +++ b/volatility/framework/symbols/windows/extensions/kdbg.py @@ -6,7 +6,7 @@ from volatility.framework import constants from volatility.framework import objects -class _KDDEBUGGER_DATA64(objects.Struct): +class _KDDEBUGGER_DATA64(objects.StructType): def get_build_lab(self): """Returns the NT build lab string from the KDBG""" diff --git a/volatility/framework/symbols/windows/extensions/pe.py b/volatility/framework/symbols/windows/extensions/pe.py index 074e1cd7d..e2dc9c852 100644 --- a/volatility/framework/symbols/windows/extensions/pe.py +++ b/volatility/framework/symbols/windows/extensions/pe.py @@ -9,7 +9,7 @@ from volatility.framework import objects, interfaces from volatility.framework.renderers import conversion -class _IMAGE_DOS_HEADER(objects.Struct): +class _IMAGE_DOS_HEADER(objects.StructType): def get_nt_header(self) -> interfaces.objects.ObjectInterface: """Carve out the NT header from this DOS header. This reflects on the @@ -146,7 +146,7 @@ class _IMAGE_DOS_HEADER(objects.Struct): counter += 1 -class _IMAGE_NT_HEADERS(objects.Struct): +class _IMAGE_NT_HEADERS(objects.StructType): def get_sections(self) -> Generator[interfaces.objects.ObjectInterface, None, None]: """Iterate through the section headers for this PE file. diff --git a/volatility/framework/symbols/windows/extensions/registry.py b/volatility/framework/symbols/windows/extensions/registry.py index 2b493a8fc..bbef73b23 100644 --- a/volatility/framework/symbols/windows/extensions/registry.py +++ b/volatility/framework/symbols/windows/extensions/registry.py @@ -59,7 +59,7 @@ class RegKeyFlags(enum.IntEnum): KEY_VIRTUAL_STORE = 0x200 -class _HMAP_ENTRY(objects.Struct): +class _HMAP_ENTRY(objects.StructType): def get_block_offset(self) -> int: try: @@ -67,8 +67,7 @@ class _HMAP_ENTRY(objects.Struct): except AttributeError: return self.BlockAddress - -class _CMHIVE(objects.Struct): +class _CMHIVE(objects.StructType): def get_name(self) -> Optional[interfaces.objects.ObjectInterface]: """Determine a name for the hive. Note that some attributes are @@ -86,7 +85,7 @@ class _CMHIVE(objects.Struct): name = property(get_name) -class _CM_KEY_BODY(objects.Struct): +class _CM_KEY_BODY(objects.StructType): """This represents an open handle to a registry key and is not tied to the registry hive file format on disk.""" @@ -119,7 +118,7 @@ class _CM_KEY_BODY(objects.Struct): return "\\".join(reversed(output)) -class _CM_KEY_NODE(objects.Struct): +class _CM_KEY_NODE(objects.StructType): """Extension to allow traversal of registry keys""" def get_volatile(self) -> bool: @@ -212,7 +211,7 @@ class _CM_KEY_NODE(objects.Struct): return reg.get_node(self.Parent).get_key_path() + '\\' + self.get_name() -class _CM_KEY_VALUE(objects.Struct): +class _CM_KEY_VALUE(objects.StructType): """Extensions to extract data from CM_KEY_VALUE nodes""" def get_name(self) -> interfaces.objects.ObjectInterface: @@ -272,10 +271,10 @@ class _CM_KEY_VALUE(objects.Struct): return output if self_type == RegValueTypes.REG_MULTI_SZ: return str(data, encoding = "utf-16-le").split("\x00")[0] - if self_type in [ - RegValueTypes.REG_BINARY, RegValueTypes.REG_FULL_RESOURCE_DESCRIPTOR, RegValueTypes.REG_RESOURCE_LIST, - RegValueTypes.REG_RESOURCE_REQUIREMENTS_LIST - ]: + if self_type in [RegValueTypes.REG_BINARY, + RegValueTypes.REG_FULL_RESOURCE_DESCRIPTOR, + RegValueTypes.REG_RESOURCE_LIST, + RegValueTypes.REG_RESOURCE_REQUIREMENTS_LIST]: return data if self_type == RegValueTypes.REG_NONE: return '' diff --git a/volatility/framework/symbols/windows/extensions/services.py b/volatility/framework/symbols/windows/extensions/services.py index 5cf452ccb..56e2a77ac 100644 --- a/volatility/framework/symbols/windows/extensions/services.py +++ b/volatility/framework/symbols/windows/extensions/services.py @@ -8,8 +8,7 @@ from volatility.framework.symbols.wrappers import Flags from volatility.framework import renderers from typing import Union - -class _SERVICE_RECORD(objects.Struct): +class _SERVICE_RECORD(objects.StructType): """A service record structure""" def is_valid(self) -> bool: @@ -44,27 +43,35 @@ class _SERVICE_RECORD(objects.Struct): # or kernel driver, the binary path is stored differently try: if "PROCESS" in self.get_type(): - return self.ServiceProcess.BinaryPath.dereference().cast( - "string", encoding = "utf-16", errors = "replace", max_length = 512) + return self.ServiceProcess.BinaryPath.dereference().cast("string", + encoding = "utf-16", + errors = "replace", + max_length = 512) else: - return self.DriverName.dereference().cast( - "string", encoding = "utf-16", errors = "replace", max_length = 512) + return self.DriverName.dereference().cast("string", + encoding = "utf-16", + errors = "replace", + max_length = 512) except exceptions.InvalidAddressException: return renderers.UnreadableValue() def get_name(self) -> Union[str, interfaces.renderers.BaseAbsentValue]: """Returns the service name""" try: - return self.ServiceName.dereference().cast( - "string", encoding = "utf-16", errors = "replace", max_length = 512) + return self.ServiceName.dereference().cast("string", + encoding = "utf-16", + errors = "replace", + max_length = 512) except exceptions.InvalidAddressException: return renderers.UnreadableValue() def get_display(self) -> Union[str, interfaces.renderers.BaseAbsentValue]: """Returns the service display""" try: - return self.DisplayName.dereference().cast( - "string", encoding = "utf-16", errors = "replace", max_length = 512) + return self.DisplayName.dereference().cast("string", + encoding = "utf-16", + errors = "replace", + max_length = 512) except exceptions.InvalidAddressException: return renderers.UnreadableValue() @@ -105,8 +112,7 @@ class _SERVICE_RECORD(objects.Struct): except exceptions.InvalidAddressException: raise StopIteration - -class _SERVICE_HEADER(objects.Struct): +class _SERVICE_HEADER(objects.StructType): """A service header structure""" def is_valid(self) -> bool: @@ -116,5 +122,7 @@ class _SERVICE_HEADER(objects.Struct): except exceptions.InvalidAddressException: return False - -class_types = {'_SERVICE_RECORD': _SERVICE_RECORD, '_SERVICE_HEADER': _SERVICE_HEADER} +class_types = { + '_SERVICE_RECORD': _SERVICE_RECORD, + '_SERVICE_HEADER': _SERVICE_HEADER +}