mirror of
https://github.com/volatilityfoundation/volatility3.git
synced 2026-08-20 13:42:23 +02:00
188 lines
10 KiB
Python
188 lines
10 KiB
Python
# This file is Copyright 2019 Volatility Foundation and licensed under the Volatility Software License 1.0
|
|
# which is available at https://www.volatilityfoundation.org/license/vsl-v1.0
|
|
#
|
|
|
|
import logging
|
|
from typing import List
|
|
|
|
from volatility.framework import interfaces, renderers, constants, symbols, exceptions
|
|
from volatility.framework.configuration import requirements
|
|
from volatility.framework.layers import scanners
|
|
from volatility.framework.renderers import format_hints
|
|
from volatility.framework.symbols import intermed
|
|
from volatility.framework.symbols.windows.extensions import services
|
|
from volatility.plugins.windows import poolscanner, vadyarascan, pslist
|
|
|
|
vollog = logging.getLogger(__name__)
|
|
|
|
|
|
class SvcScan(interfaces.plugins.PluginInterface):
|
|
"""Scans for windows services."""
|
|
|
|
_version = (1, 0, 0)
|
|
|
|
is_vista_or_later = poolscanner.os_distinguisher(version_check = lambda x: x >= (6, 0),
|
|
fallback_checks = [("KdCopyDataBlock", None, True)])
|
|
|
|
is_windows_xp = poolscanner.os_distinguisher(version_check = lambda x: (5, 1) <= x < (5, 2),
|
|
fallback_checks = [("KdCopyDataBlock", None, False),
|
|
("_HANDLE_TABLE", "HandleCount", True)])
|
|
|
|
is_xp_or_2003 = poolscanner.os_distinguisher(version_check = lambda x: (5, 1) <= x < (6, 0),
|
|
fallback_checks = [("KdCopyDataBlock", None, False),
|
|
("_HANDLE_TABLE", "HandleCount", True)])
|
|
|
|
is_win10_up_to_15063 = poolscanner.os_distinguisher(version_check = lambda x: (10, 0) <= x < (10, 0, 16299),
|
|
fallback_checks = [("ObHeaderCookie", None, True),
|
|
("_HANDLE_TABLE", "HandleCount", False),
|
|
("ObHeaderCookie", None, True),
|
|
("_EPROCESS", "ControlFlowGuardEnabled", True)])
|
|
|
|
is_win10_16299_or_later = poolscanner.os_distinguisher(version_check = lambda x: x >= (10, 0, 16299),
|
|
fallback_checks = [("ObHeaderCookie", None, True),
|
|
("_HANDLE_TABLE", "HandleCount", False),
|
|
("ObHeaderCookie", None, True)])
|
|
|
|
@classmethod
|
|
def get_requirements(cls) -> List[interfaces.configuration.RequirementInterface]:
|
|
# Since we're calling the plugin, make sure we have the plugin's requirements
|
|
return [
|
|
requirements.TranslationLayerRequirement(name = 'primary',
|
|
description = 'Memory layer for the kernel',
|
|
architectures = ["Intel32", "Intel64"]),
|
|
requirements.SymbolTableRequirement(name = "nt_symbols", description = "Windows kernel symbols"),
|
|
requirements.PluginRequirement(name = 'pslist', plugin = pslist.PsList, version = (1, 0, 0)),
|
|
requirements.PluginRequirement(name = 'poolscanner', plugin = poolscanner.PoolScanner, version = (1, 0, 0)),
|
|
requirements.PluginRequirement(name = 'vadyarascan', plugin = vadyarascan.VadYaraScan, version = (1, 0, 0))
|
|
]
|
|
|
|
@staticmethod
|
|
def get_record_tuple(service_record: interfaces.objects.ObjectInterface):
|
|
return (format_hints.Hex(service_record.vol.offset), service_record.Order, service_record.get_pid(),
|
|
service_record.Start.description, service_record.State.description, service_record.get_type(),
|
|
service_record.get_name(), service_record.get_display(), service_record.get_binary())
|
|
|
|
@staticmethod
|
|
def create_service_table(context: interfaces.context.ContextInterface, symbol_table: str, config_path: str) -> str:
|
|
"""Constructs a symbol table containing the symbols for services
|
|
depending upon the operating system in use.
|
|
|
|
Args:
|
|
context: The context to retrieve required elements (layers, symbol tables) from
|
|
symbol_table: The name of the table containing the kernel symbols
|
|
config_path: The configuration path for any settings required by the new table
|
|
|
|
Returns:
|
|
A symbol table containing the symbols necessary for services
|
|
"""
|
|
native_types = context.symbol_space[symbol_table].natives
|
|
is_64bit = symbols.symbol_table_is_64bit(context, symbol_table)
|
|
|
|
if SvcScan.is_windows_xp(context = context, symbol_table = symbol_table) and not is_64bit:
|
|
symbol_filename = "services-xp-x86"
|
|
elif SvcScan.is_xp_or_2003(context = context, symbol_table = symbol_table) and is_64bit:
|
|
symbol_filename = "services-xp-2003-x64"
|
|
elif SvcScan.is_win10_16299_or_later(context = context, symbol_table = symbol_table) and is_64bit:
|
|
symbol_filename = "services-win10-16299-x64"
|
|
elif SvcScan.is_win10_16299_or_later(context = context, symbol_table = symbol_table) and not is_64bit:
|
|
symbol_filename = "services-win10-16299-x86"
|
|
elif SvcScan.is_win10_up_to_15063(context = context, symbol_table = symbol_table) and is_64bit:
|
|
symbol_filename = "services-win10-15063-x64"
|
|
elif SvcScan.is_win10_up_to_15063(context = context, symbol_table = symbol_table) and not is_64bit:
|
|
symbol_filename = "services-win10-15063-x86"
|
|
elif poolscanner.PoolScanner.is_windows_8_or_later(context = context, symbol_table = symbol_table) and is_64bit:
|
|
symbol_filename = "services-win8-x64"
|
|
elif poolscanner.PoolScanner.is_windows_8_or_later(context = context,
|
|
symbol_table = symbol_table) and not is_64bit:
|
|
symbol_filename = "services-win8-x86"
|
|
elif SvcScan.is_vista_or_later(context = context, symbol_table = symbol_table) and is_64bit:
|
|
symbol_filename = "services-vista-x64"
|
|
elif SvcScan.is_vista_or_later(context = context, symbol_table = symbol_table) and not is_64bit:
|
|
symbol_filename = "services-vista-x86"
|
|
else:
|
|
raise NotImplementedError("This version of Windows is not supported!")
|
|
|
|
return intermed.IntermediateSymbolTable.create(context,
|
|
config_path,
|
|
"windows",
|
|
symbol_filename,
|
|
class_types = services.class_types,
|
|
native_types = native_types)
|
|
|
|
def _generator(self):
|
|
|
|
service_table_name = self.create_service_table(self.context, self.config["nt_symbols"], self.config_path)
|
|
|
|
relative_tag_offset = self.context.symbol_space.get_type(service_table_name + constants.BANG +
|
|
"_SERVICE_RECORD").relative_child_offset("Tag")
|
|
|
|
filter_func = pslist.PsList.create_name_filter(["services.exe"])
|
|
|
|
is_vista_or_later = SvcScan.is_vista_or_later(context = self.context, symbol_table = self.config["nt_symbols"])
|
|
|
|
if is_vista_or_later:
|
|
service_tag = b"serH"
|
|
else:
|
|
service_tag = b"sErv"
|
|
|
|
seen = []
|
|
|
|
for task in pslist.PsList.list_processes(context = self.context,
|
|
layer_name = self.config['primary'],
|
|
symbol_table = self.config['nt_symbols'],
|
|
filter_func = filter_func):
|
|
|
|
proc_id = "Unknown"
|
|
try:
|
|
proc_id = task.UniqueProcessId
|
|
proc_layer_name = task.add_process_layer()
|
|
except exceptions.InvalidAddressException as excp:
|
|
vollog.debug("Process {}: invalid address {} in layer {}".format(proc_id, excp.invalid_address,
|
|
excp.layer_name))
|
|
continue
|
|
|
|
layer = self.context.layers[proc_layer_name]
|
|
|
|
for offset in layer.scan(context = self.context,
|
|
scanner = scanners.BytesScanner(needle = service_tag),
|
|
sections = vadyarascan.VadYaraScan.get_vad_maps(task)):
|
|
|
|
if not is_vista_or_later:
|
|
service_record = self.context.object(service_table_name + constants.BANG + "_SERVICE_RECORD",
|
|
offset = offset - relative_tag_offset,
|
|
layer_name = proc_layer_name)
|
|
|
|
if not service_record.is_valid():
|
|
continue
|
|
|
|
yield (0, self.get_record_tuple(service_record))
|
|
else:
|
|
service_header = self.context.object(service_table_name + constants.BANG + "_SERVICE_HEADER",
|
|
offset = offset,
|
|
layer_name = proc_layer_name)
|
|
|
|
if not service_header.is_valid():
|
|
continue
|
|
|
|
# since we walk the s-list backwards, if we've seen
|
|
# an object, then we've also seen all objects that
|
|
# exist before it, thus we can break at that time.
|
|
for service_record in service_header.ServiceRecord.traverse():
|
|
if service_record in seen:
|
|
break
|
|
seen.append(service_record)
|
|
yield (0, self.get_record_tuple(service_record))
|
|
|
|
def run(self):
|
|
return renderers.TreeGrid([
|
|
('Offset', format_hints.Hex),
|
|
('Order', int),
|
|
('Pid', int),
|
|
('Start', str),
|
|
('State', str),
|
|
('Type', str),
|
|
('Name', str),
|
|
('Display', str),
|
|
('Binary', str),
|
|
], self._generator())
|