Yapf reformat all files.

This commit is contained in:
Mike Auty
2019-03-04 22:11:44 +00:00
parent 6613c7db19
commit 7904be3fd2
10 changed files with 79 additions and 152 deletions
@@ -18,45 +18,26 @@
# specific language governing rights and limitations under the License.
#
from volatility.plugins.windows.driverscan import DriverScan
import volatility.framework.interfaces.plugins as plugins
from volatility.framework import constants
from volatility.framework import renderers, exceptions
from volatility.framework.configuration import requirements
from volatility.framework.renderers import format_hints
from volatility.plugins.windows.driverscan import DriverScan
from volatility.plugins.windows import ssdt
from volatility.framework import constants
MAJOR_FUNCTIONS = [
'IRP_MJ_CREATE',
'IRP_MJ_CREATE_NAMED_PIPE',
'IRP_MJ_CLOSE',
'IRP_MJ_READ',
'IRP_MJ_WRITE',
'IRP_MJ_QUERY_INFORMATION',
'IRP_MJ_SET_INFORMATION',
'IRP_MJ_QUERY_EA',
'IRP_MJ_SET_EA',
'IRP_MJ_FLUSH_BUFFERS',
'IRP_MJ_QUERY_VOLUME_INFORMATION',
'IRP_MJ_SET_VOLUME_INFORMATION',
'IRP_MJ_DIRECTORY_CONTROL',
'IRP_MJ_FILE_SYSTEM_CONTROL',
'IRP_MJ_DEVICE_CONTROL',
'IRP_MJ_INTERNAL_DEVICE_CONTROL',
'IRP_MJ_SHUTDOWN',
'IRP_MJ_LOCK_CONTROL',
'IRP_MJ_CLEANUP',
'IRP_MJ_CREATE_MAILSLOT',
'IRP_MJ_QUERY_SECURITY',
'IRP_MJ_SET_SECURITY',
'IRP_MJ_POWER',
'IRP_MJ_SYSTEM_CONTROL',
'IRP_MJ_DEVICE_CHANGE',
'IRP_MJ_QUERY_QUOTA',
'IRP_MJ_SET_QUOTA',
'IRP_MJ_CREATE', 'IRP_MJ_CREATE_NAMED_PIPE', 'IRP_MJ_CLOSE', 'IRP_MJ_READ', 'IRP_MJ_WRITE',
'IRP_MJ_QUERY_INFORMATION', 'IRP_MJ_SET_INFORMATION', 'IRP_MJ_QUERY_EA', 'IRP_MJ_SET_EA', 'IRP_MJ_FLUSH_BUFFERS',
'IRP_MJ_QUERY_VOLUME_INFORMATION', 'IRP_MJ_SET_VOLUME_INFORMATION', 'IRP_MJ_DIRECTORY_CONTROL',
'IRP_MJ_FILE_SYSTEM_CONTROL', 'IRP_MJ_DEVICE_CONTROL', 'IRP_MJ_INTERNAL_DEVICE_CONTROL', 'IRP_MJ_SHUTDOWN',
'IRP_MJ_LOCK_CONTROL', 'IRP_MJ_CLEANUP', 'IRP_MJ_CREATE_MAILSLOT', 'IRP_MJ_QUERY_SECURITY', 'IRP_MJ_SET_SECURITY',
'IRP_MJ_POWER', 'IRP_MJ_SYSTEM_CONTROL', 'IRP_MJ_DEVICE_CHANGE', 'IRP_MJ_QUERY_QUOTA', 'IRP_MJ_SET_QUOTA',
'IRP_MJ_PNP'
]
class DriverIrp(plugins.PluginInterface):
"""List IRPs for drivers in a particular windows memory image"""
@@ -70,14 +51,9 @@ class DriverIrp(plugins.PluginInterface):
def _generator(self):
collection = ssdt.SSDT.build_module_collection(self.context,
self.config['primary'],
self.config['nt_symbols'])
collection = ssdt.SSDT.build_module_collection(self.context, self.config['primary'], self.config['nt_symbols'])
for driver in DriverScan.scan_drivers(self.context,
self.config['primary'],
self.config['nt_symbols']):
for driver in DriverScan.scan_drivers(self.context, self.config['primary'], self.config['nt_symbols']):
try:
driver_name = driver.get_driver_name()
@@ -92,26 +68,19 @@ class DriverIrp(plugins.PluginInterface):
for symbol in symbol_generator:
symbols_found = True
yield (0, (format_hints.Hex(driver.vol.offset),
driver_name,
MAJOR_FUNCTIONS[i],
format_hints.Hex(address),
module_name,
symbol.split(constants.BANG)[1]))
yield (0, (format_hints.Hex(driver.vol.offset), driver_name, MAJOR_FUNCTIONS[i],
format_hints.Hex(address), module_name, symbol.split(constants.BANG)[1]))
if not symbols_found:
yield (0, (format_hints.Hex(driver.vol.offset),
driver_name,
MAJOR_FUNCTIONS[i],
format_hints.Hex(address),
module_name,
renderers.NotAvailableValue()))
yield (0, (format_hints.Hex(driver.vol.offset), driver_name, MAJOR_FUNCTIONS[i],
format_hints.Hex(address), module_name, renderers.NotAvailableValue()))
def run(self):
return renderers.TreeGrid([("Offset", format_hints.Hex),
("Driver Name", str),
("IRP", str),
("Address", format_hints.Hex),
("Module", str),
("Symbol", str),],
self._generator())
return renderers.TreeGrid([
("Offset", format_hints.Hex),
("Driver Name", str),
("IRP", str),
("Address", format_hints.Hex),
("Module", str),
("Symbol", str),
], self._generator())
@@ -20,11 +20,13 @@
from typing import Iterable
import volatility.plugins.windows.poolscanner as poolscanner
import volatility.framework.interfaces.plugins as plugins
from volatility.framework import renderers, interfaces, exceptions
from volatility.framework.configuration import requirements
from volatility.framework.renderers import format_hints
import volatility.plugins.windows.poolscanner as poolscanner
class DriverScan(plugins.PluginInterface):
"""Scans for drivers present in a particular windows memory image"""
@@ -45,21 +47,15 @@ class DriverScan(plugins.PluginInterface):
Iterable[interfaces.objects.ObjectInterface]:
"""Scans for drivers using the poolscanner module and constraints"""
constraints = poolscanner.PoolScanner.builtin_constraints(symbol_table,
[b'Dri\xf6', b'Driv'])
constraints = poolscanner.PoolScanner.builtin_constraints(symbol_table, [b'Dri\xf6', b'Driv'])
for result in poolscanner.PoolScanner.generate_pool_scan(context,
layer_name,
symbol_table,
constraints):
for result in poolscanner.PoolScanner.generate_pool_scan(context, layer_name, symbol_table, constraints):
_constraint, mem_object, _header = result
yield mem_object
def _generator(self):
for driver in self.scan_drivers(self.context,
self.config['primary'],
self.config['nt_symbols']):
for driver in self.scan_drivers(self.context, self.config['primary'], self.config['nt_symbols']):
try:
driver_name = driver.get_driver_name()
@@ -76,18 +72,10 @@ class DriverScan(plugins.PluginInterface):
except exceptions.InvalidAddressException:
name = renderers.NotApplicableValue()
yield (0, (format_hints.Hex(driver.vol.offset),
format_hints.Hex(driver.DriverStart),
format_hints.Hex(driver.DriverSize),
service_key,
driver_name,
name))
yield (0, (format_hints.Hex(driver.vol.offset), format_hints.Hex(driver.DriverStart),
format_hints.Hex(driver.DriverSize), service_key, driver_name, name))
def run(self):
return renderers.TreeGrid([("Offset", format_hints.Hex),
("Start", format_hints.Hex),
("Size", format_hints.Hex),
("Service Key", str),
("Driver Name", str),
("Name", str)],
self._generator())
("Start", format_hints.Hex), ("Size", format_hints.Hex), ("Service Key", str),
("Driver Name", str), ("Name", str)], self._generator())
@@ -26,6 +26,7 @@ from volatility.framework.configuration import requirements
from volatility.framework.renderers import format_hints
import volatility.plugins.windows.poolscanner as poolscanner
class FileScan(plugins.PluginInterface):
"""Scans for file objects present in a particular windows memory image"""
@@ -45,21 +46,15 @@ class FileScan(plugins.PluginInterface):
Iterable[interfaces.objects.ObjectInterface]:
"""Scans for file objects using the poolscanner module and constraints"""
constraints = poolscanner.PoolScanner.builtin_constraints(symbol_table,
[b'Fil\xe5', b'File'])
constraints = poolscanner.PoolScanner.builtin_constraints(symbol_table, [b'Fil\xe5', b'File'])
for result in poolscanner.PoolScanner.generate_pool_scan(context,
layer_name,
symbol_table,
constraints):
for result in poolscanner.PoolScanner.generate_pool_scan(context, layer_name, symbol_table, constraints):
_constraint, mem_object, _header = result
yield mem_object
def _generator(self):
for fileobj in self.scan_files(self.context,
self.config['primary'],
self.config['nt_symbols']):
for fileobj in self.scan_files(self.context, self.config['primary'], self.config['nt_symbols']):
try:
file_name = fileobj.FileName.String
@@ -69,5 +64,7 @@ class FileScan(plugins.PluginInterface):
yield (0, (format_hints.Hex(fileobj.vol.offset), file_name))
def run(self):
return renderers.TreeGrid([("Offset", format_hints.Hex), ("Name", str),],
self._generator())
return renderers.TreeGrid([
("Offset", format_hints.Hex),
("Name", str),
], self._generator())
@@ -306,24 +306,17 @@ class Handles(interfaces_plugins.PluginInterface):
"Cannot access _OBJECT_HEADER at {0:#x}".format(entry.vol.offset))
continue
yield (0, (proc.UniqueProcessId,
process_name,
format_hints.Hex(entry.Body.vol.offset),
format_hints.Hex(entry.HandleValue),
obj_type,
format_hints.Hex(entry.GrantedAccess),
yield (0, (proc.UniqueProcessId, process_name, format_hints.Hex(entry.Body.vol.offset),
format_hints.Hex(entry.HandleValue), obj_type, format_hints.Hex(entry.GrantedAccess),
obj_name))
def run(self):
filter_func = pslist.PsList.create_filter([self.config.get('pid', None)])
return renderers.TreeGrid([("PID", int), ("Process", str),
("Offset", format_hints.Hex),
("HandleValue", format_hints.Hex),
("Type", str),
("GrantedAccess", format_hints.Hex),
("Name", str)],
return renderers.TreeGrid([("PID", int), ("Process", str), ("Offset", format_hints.Hex),
("HandleValue", format_hints.Hex), ("Type", str),
("GrantedAccess", format_hints.Hex), ("Name", str)],
self._generator(
pslist.PsList.list_processes(
self.context,
@@ -20,11 +20,13 @@
from typing import Iterable
import volatility.plugins.windows.poolscanner as poolscanner
import volatility.framework.interfaces.plugins as plugins
from volatility.framework import renderers, interfaces, exceptions
from volatility.framework.configuration import requirements
from volatility.framework.renderers import format_hints
import volatility.plugins.windows.poolscanner as poolscanner
class ModScan(plugins.PluginInterface):
"""Scans for modules present in a particular windows memory image"""
@@ -47,18 +49,13 @@ class ModScan(plugins.PluginInterface):
constraints = poolscanner.PoolScanner.builtin_constraints(symbol_table, [b'MmLd'])
for result in poolscanner.PoolScanner.generate_pool_scan(context,
layer_name,
symbol_table,
constraints):
for result in poolscanner.PoolScanner.generate_pool_scan(context, layer_name, symbol_table, constraints):
_constraint, mem_object, _header = result
yield mem_object
def _generator(self):
for mod in self.scan_modules(self.context,
self.config['primary'],
self.config['nt_symbols']):
for mod in self.scan_modules(self.context, self.config['primary'], self.config['nt_symbols']):
try:
BaseDllName = mod.BaseDllName.get_string()
@@ -79,9 +76,5 @@ class ModScan(plugins.PluginInterface):
))
def run(self):
return renderers.TreeGrid([("Offset", format_hints.Hex),
("Base", format_hints.Hex),
("Size", format_hints.Hex),
("Name", str),
("Path", str)],
self._generator())
return renderers.TreeGrid([("Offset", format_hints.Hex), ("Base", format_hints.Hex), ("Size", format_hints.Hex),
("Name", str), ("Path", str)], self._generator())
@@ -26,6 +26,7 @@ from volatility.framework.configuration import requirements
from volatility.framework.renderers import format_hints
import volatility.plugins.windows.poolscanner as poolscanner
class MutantScan(plugins.PluginInterface):
"""Scans for mutexes present in a particular windows memory image"""
@@ -45,21 +46,15 @@ class MutantScan(plugins.PluginInterface):
Iterable[interfaces.objects.ObjectInterface]:
"""Scans for mutants using the poolscanner module and constraints"""
constraints = poolscanner.PoolScanner.builtin_constraints(symbol_table,
[b'Mut\xe1', b'Muta'])
constraints = poolscanner.PoolScanner.builtin_constraints(symbol_table, [b'Mut\xe1', b'Muta'])
for result in poolscanner.PoolScanner.generate_pool_scan(context,
layer_name,
symbol_table,
constraints):
for result in poolscanner.PoolScanner.generate_pool_scan(context, layer_name, symbol_table, constraints):
_constraint, mem_object, _header = result
yield mem_object
def _generator(self):
for mutant in self.scan_mutants(self.context,
self.config['primary'],
self.config['nt_symbols']):
for mutant in self.scan_mutants(self.context, self.config['primary'], self.config['nt_symbols']):
try:
name = mutant.get_name()
@@ -69,5 +64,7 @@ class MutantScan(plugins.PluginInterface):
yield (0, (format_hints.Hex(mutant.vol.offset), name))
def run(self):
return renderers.TreeGrid([("Offset", format_hints.Hex), ("Name", str),],
self._generator())
return renderers.TreeGrid([
("Offset", format_hints.Hex),
("Name", str),
], self._generator())
+8 -15
View File
@@ -28,6 +28,7 @@ from volatility.framework.renderers import format_hints
from volatility.plugins import timeliner
import volatility.plugins.windows.poolscanner as poolscanner
class PsScan(plugins.PluginInterface, timeliner.TimeLinerInterface):
"""Scans for processes present in a particular windows memory image"""
@@ -47,27 +48,20 @@ class PsScan(plugins.PluginInterface, timeliner.TimeLinerInterface):
Iterable[interfaces.objects.ObjectInterface]:
"""Scans for processes using the poolscanner module and constraints"""
constraints = poolscanner.PoolScanner.builtin_constraints(symbol_table,
[b'Pro\xe3', b'Proc'])
constraints = poolscanner.PoolScanner.builtin_constraints(symbol_table, [b'Pro\xe3', b'Proc'])
for result in poolscanner.PoolScanner.generate_pool_scan(context,
layer_name,
symbol_table,
constraints):
for result in poolscanner.PoolScanner.generate_pool_scan(context, layer_name, symbol_table, constraints):
_constraint, mem_object, _header = result
yield mem_object
def _generator(self):
for proc in self.scan_processes(
self.context,
self.config['primary'],
self.config['nt_symbols']):
for proc in self.scan_processes(self.context, self.config['primary'], self.config['nt_symbols']):
yield (0, (proc.UniqueProcessId, proc.InheritedFromUniqueProcessId,
proc.ImageFileName.cast("string", max_length = proc.ImageFileName.vol.count, errors = 'replace'),
format_hints.Hex(proc.vol.offset), proc.ActiveThreads, proc.get_handle_count(), proc.get_session_id(),
proc.get_is_wow64(), proc.get_create_time(), proc.get_exit_time()))
format_hints.Hex(proc.vol.offset), proc.ActiveThreads, proc.get_handle_count(),
proc.get_session_id(), proc.get_is_wow64(), proc.get_create_time(), proc.get_exit_time()))
def generate_timeline(self):
for row in self._generator():
@@ -77,8 +71,7 @@ class PsScan(plugins.PluginInterface, timeliner.TimeLinerInterface):
yield (description, timeliner.TimeLinerType.MODIFIED, row_data[9])
def run(self):
return renderers.TreeGrid([("PID", int), ("PPID", int), ("ImageFileName", str),
("Offset", format_hints.Hex), ("Threads", int),
("Handles", int), ("SessionId", int), ("Wow64", bool),
return renderers.TreeGrid([("PID", int), ("PPID", int), ("ImageFileName", str), ("Offset", format_hints.Hex),
("Threads", int), ("Handles", int), ("SessionId", int), ("Wow64", bool),
("CreateTime", datetime.datetime), ("ExitTime", datetime.datetime)],
self._generator())
+3 -11
View File
@@ -44,9 +44,7 @@ class SSDT(plugins.PluginInterface):
]
@classmethod
def build_module_collection(cls,
context: interfaces.context.ContextInterface,
layer_name: str,
def build_module_collection(cls, context: interfaces.context.ContextInterface, layer_name: str,
symbol_table: str) -> contexts.ModuleCollection:
"""Builds a collection of modules"""
@@ -68,11 +66,7 @@ class SSDT(plugins.PluginInterface):
else:
symbol_table_name = None
context_module = contexts.SizedModule(context,
module_name,
layer_name,
mod.DllBase,
mod.SizeOfImage,
context_module = contexts.SizedModule(context, module_name, layer_name, mod.DllBase, mod.SizeOfImage,
symbol_table_name)
context_modules.append(context_module)
@@ -82,9 +76,7 @@ class SSDT(plugins.PluginInterface):
def _generator(self) -> Iterator[Tuple[int, Tuple[int, int, Any, Any]]]:
layer_name = self.config['primary']
collection = self.build_module_collection(self.context,
self.config["primary"],
self.config["nt_symbols"])
collection = self.build_module_collection(self.context, self.config["primary"], self.config["nt_symbols"])
kvo = self.context.memory[layer_name].config['kernel_virtual_offset']
ntkrnlmp = self.context.module(self.config["nt_symbols"], layer_name = layer_name, offset = kvo)
@@ -239,9 +239,10 @@ class vm_area_struct(objects.Struct):
elif flags_str == "r-x" and self.vm_file.dereference().vol.offset == 0:
ret = True
return ret
class qstr(objects.Struct):
def name_as_str(self) -> str:
@@ -436,6 +436,7 @@ class _DEVICE_OBJECT(objects.Struct, ExecutiveObject):
header = self.object_header()
return header.NameInfo.Name.String # type: ignore
class _DRIVER_OBJECT(objects.Struct, ExecutiveObject):
"""A class for kernel driver objects."""
@@ -447,6 +448,7 @@ class _DRIVER_OBJECT(objects.Struct, ExecutiveObject):
"""Determine if the object is valid"""
return True
class _FILE_OBJECT(objects.Struct, ExecutiveObject):
"""A class for windows file objects"""
@@ -467,6 +469,7 @@ class _FILE_OBJECT(objects.Struct, ExecutiveObject):
return name
class _KMUTANT(objects.Struct, ExecutiveObject):
"""A class for windows mutant objects"""
@@ -479,6 +482,7 @@ class _KMUTANT(objects.Struct, ExecutiveObject):
header = self.object_header()
return header.NameInfo.Name.String # type: ignore
class _OBJECT_HEADER(objects.Struct):
"""A class for the headers for executive kernel objects, which contains
quota information, ownership details, naming data, and ACLs."""