WEBDIENSTE/volatility3
1
0
Fork 0
mirror of https://github.com/volatilityfoundation/volatility3.git synced 2026-09-23 18:14:51 +02:00
Code Issues Packages Projects Releases Wiki Activity

refactor poolscanner to be leveraged by other plugins. add psscan

Browse Source
This commit is contained in:
Michael Ligh
2019-02-07 09:08:32 +00:00
committed by ikelos
parent f898dca296
commit c1341e6278
2 changed files with 123 additions and 28 deletions
Download Patch File Download Diff File Expand all files Collapse all files
volatility/framework/plugins/windows/poolscanner.py
+42 -28
View File
@@ -139,38 +139,65 @@ class PoolScanner(plugins.PluginInterface):
and not PoolScanner.is_windows_8_or_later(context, layer_name, symbol_table))
def _generator(self):
for result in self.generate_pool_scan(self.context,
self.config["primary"],
self.config["nt_symbols"]):
constraint, mem_object, header = result
# generate some type-specific info for sanity checking
if constraint.object_type == "Process":
name = mem_object.ImageFileName.cast(
"string", max_length=mem_object.ImageFileName.vol.count, errors="replace")
elif constraint.object_type == "File":
try:
name = mem_object.FileName.String
except exceptions.PagedInvalidAddressException:
vollog.log(constants.LOGLEVEL_VVV, "Skipping file at {0:#x}".format(mem_object.vol.offset))
continue
else:
name = renderers.NotApplicableValue()
yield (0, (constraint.type_name, format_hints.Hex(header.vol.offset), header.vol.layer_name, name))
@classmethod
def generate_pool_scan(cls,
context: interfaces.context.ContextInterface,
layer_name: str,
symbol_table: str) \
-> Generator[Tuple[PoolConstraint, interfaces.objects.ObjectInterface, interfaces.objects.ObjectInterface], None, None]:
constraints = [
# atom tables
PoolConstraint(
b'AtmT',
type_name = self.config["nt_symbols"] + constants.BANG + "_RTL_ATOM_TABLE",
type_name = symbol_table + constants.BANG + "_RTL_ATOM_TABLE",
size = (200, None),
page_type = PoolType.PAGED | PoolType.NONPAGED | PoolType.FREE),
# processes on windows before windows 8
PoolConstraint(
b'Pro\xe3',
type_name = self.config["nt_symbols"] + constants.BANG + "_EPROCESS",
type_name = symbol_table + constants.BANG + "_EPROCESS",
object_type = "Process",
size = (600, None),
page_type = PoolType.PAGED | PoolType.NONPAGED | PoolType.FREE),
# processes on windows starting with windows 8
PoolConstraint(
b'Proc',
type_name = self.config["nt_symbols"] + constants.BANG + "_EPROCESS",
type_name = symbol_table + constants.BANG + "_EPROCESS",
object_type = "Process",
size = (600, None),
page_type = PoolType.PAGED | PoolType.NONPAGED | PoolType.FREE),
# files on windows before windows 8
PoolConstraint(
b'Fil\xe5',
type_name = self.config["nt_symbols"] + constants.BANG + "_FILE_OBJECT",
type_name = symbol_table + constants.BANG + "_FILE_OBJECT",
object_type = "File",
size = (150, None),
page_type = PoolType.PAGED | PoolType.NONPAGED | PoolType.FREE),
# files on windows starting with windows 8
PoolConstraint(
b'File',
type_name = self.config["nt_symbols"] + constants.BANG + "_FILE_OBJECT",
type_name = symbol_table + constants.BANG + "_FILE_OBJECT",
object_type = "File",
size = (150, None),
page_type = PoolType.PAGED | PoolType.NONPAGED | PoolType.FREE),
@@ -178,24 +205,24 @@ class PoolScanner(plugins.PluginInterface):
# get the object type map
type_map = handles.Handles.list_objects(
context = self.context, layer_name = self.config["primary"], symbol_table = self.config["nt_symbols"])
context = context, layer_name = layer_name, symbol_table = symbol_table)
cookie = handles.Handles.find_cookie(
context = self.context, layer_name = self.config["primary"], symbol_table = self.config["nt_symbols"])
context = context, layer_name = layer_name, symbol_table = symbol_table)
is_windows_10 = self.is_windows_10(context = self._context, symbol_table = self.config["nt_symbols"])
is_windows_8_or_later = self.is_windows_8_or_later(
context = self._context, layer_name = self.config["primary"], symbol_table = self.config["nt_symbols"])
is_windows_10 = cls.is_windows_10(context = context, symbol_table = symbol_table)
is_windows_8_or_later = cls.is_windows_8_or_later(
context = context, layer_name = layer_name, symbol_table = symbol_table)
# start off with the primary virtual layer
scan_layer = self.config['primary']
scan_layer = layer_name
# switch to a non-virtual layer if necessary
if not is_windows_10:
scan_layer = self.context.memory[scan_layer].config['memory_layer']
scan_layer = context.memory[scan_layer].config['memory_layer']
for constraint, header in self.pool_scan(
self._context, scan_layer, self.config['nt_symbols'], constraints, alignment = 8):
for constraint, header in cls.pool_scan(
context, scan_layer, symbol_table, constraints, alignment = 8):
mem_object = header.get_object(
type_name = constraint.type_name,
@@ -209,20 +236,7 @@ class PoolScanner(plugins.PluginInterface):
vollog.log(constants.LOGLEVEL_VVV, "Cannot create an instance of {}".format(constraint.type_name))
continue
# generate some type-specific info for sanity checking
if constraint.object_type == "Process":
name = mem_object.ImageFileName.cast(
"string", max_length = mem_object.ImageFileName.vol.count, errors = "replace")
elif constraint.object_type == "File":
try:
name = mem_object.FileName.String
except exceptions.PagedInvalidAddressException:
vollog.log(constants.LOGLEVEL_VVV, "Skipping file at {0:#x}".format(mem_object.vol.offset))
continue
else:
name = renderers.NotApplicableValue()
yield (0, (constraint.type_name, format_hints.Hex(header.vol.offset), header.vol.layer_name, name))
yield constraint, mem_object, header
@classmethod
def pool_scan(cls,
volatility/framework/plugins/windows/psscan.py
+81
View File
@@ -0,0 +1,81 @@
# This file was contributed to the Volatility Framework Version 3.
# Copyright (C) 2018 Volatility Foundation.
#
# THE LICENSED WORK IS PROVIDED UNDER THE TERMS OF THE Volatility Contributors
# Public License V1.0("LICENSE") AS FIRST COMPLETED BY: Volatility Foundation,
# Inc. ANY USE, PUBLIC DISPLAY, PUBLIC PERFORMANCE, REPRODUCTION OR DISTRIBUTION
# OF, OR PREPARATION OF SUBSEQUENT WORKS, DERIVATIVE WORKS OR DERIVED WORKS BASED
# ON, THE LICENSED WORK CONSTITUTES RECIPIENT'S ACCEPTANCE OF THIS LICENSE AND ITS
# TERMS, WHETHER OR NOT SUCH RECIPIENT READS THE TERMS OF THE LICENSE. "LICENSED
# WORK,” “RECIPIENT" AND “DISTRIBUTOR" ARE DEFINED IN THE LICENSE. A COPY OF THE
# LICENSE IS LOCATED IN THE TEXT FILE ENTITLED "LICENSE.txt" ACCOMPANYING THE
# CONTENTS OF THIS FILE. IF A COPY OF THE LICENSE DOES NOT ACCOMPANY THIS FILE, A
# COPY OF THE LICENSE MAY ALSO BE OBTAINED AT THE FOLLOWING WEB SITE:
# https://www.volatilityfoundation.org/license/vcpl_v1.0
#
# Software distributed under the License is distributed on an "AS IS" basis,
# WITHOUT WARRANTY OF ANY KIND, either express or implied. See the License for the
# specific language governing rights and limitations under the License.
#
import datetime
from typing import Iterable
import volatility.framework.interfaces.plugins as plugins
from volatility.framework import renderers, interfaces
from volatility.framework.configuration import requirements
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"""
@classmethod
def get_requirements(cls):
return [
requirements.TranslationLayerRequirement(
name = 'primary', description = 'Memory layer for the kernel', architectures = ["Intel32", "Intel64"]),
requirements.SymbolRequirement(name = "nt_symbols", description = "Windows kernel symbols"),
]
@classmethod
def scan_processes(cls,
context: interfaces.context.ContextInterface,
layer_name: str,
symbol_table: str) -> \
Iterable[interfaces.objects.ObjectInterface]:
"""Scans for processes using the poolscanner module and constraints"""
for result in poolscanner.PoolScanner.generate_pool_scan(context,
layer_name,
symbol_table):
constraint, mem_object, _header = result
if constraint.object_type == "Process":
yield mem_object
def _generator(self):
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()))
def generate_timeline(self):
for row in self._generator():
_depth, row_data = row
description = "Process: {} ({})".format(row_data[2], row_data[3])
yield (description, timeliner.TimeLinerType.CREATED, row_data[8])
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),
("CreateTime", datetime.datetime), ("ExitTime", datetime.datetime)],
self._generator())