mirror of
https://github.com/volatilityfoundation/volatility3.git
synced 2026-08-17 20:35:40 +02:00
Merge pull request #1833 from SolitudePy/psxview_malware
Malware categorization: windows.psxview
This commit is contained in:
ikelos
@@ -0,0 +1,241 @@
|
||||
import datetime
|
||||
import logging
|
||||
import string
|
||||
from itertools import chain
|
||||
from typing import Dict, Iterable, List
|
||||
|
||||
from volatility3.framework import constants, exceptions, renderers
|
||||
from volatility3.framework.configuration import requirements
|
||||
from volatility3.framework.interfaces import plugins
|
||||
from volatility3.framework.renderers import format_hints
|
||||
from volatility3.framework.symbols.windows import extensions
|
||||
from volatility3.plugins.windows import handles, pslist, psscan, thrdscan
|
||||
|
||||
vollog = logging.getLogger(__name__)
|
||||
|
||||
|
||||
class PsXView(plugins.PluginInterface):
|
||||
"""Lists all processes found via four of the methods described in \"The Art of Memory Forensics\" which may help \
|
||||
identify processes that are trying to hide themselves.
|
||||
|
||||
We recommend using -r pretty if you are looking at this plugin's output in a terminal."""
|
||||
|
||||
# I've omitted the desktop thread scanning method because Volatility3 doesn't appear to have the functionality
|
||||
# which the original plugin used to do it.
|
||||
|
||||
# The sessions method is omitted because it begins with the list of processes found by Pslist anyway.
|
||||
|
||||
# Lastly, I've omitted the pspcid method because I could not for the life of me get it to work. I saved the
|
||||
# code I do have from it, and will happily share it if anyone else wants to add it.
|
||||
|
||||
_required_framework_version = (2, 0, 0)
|
||||
_version = (1, 0, 0)
|
||||
|
||||
valid_proc_name_chars = set(
|
||||
string.ascii_lowercase + string.ascii_uppercase + "." + " "
|
||||
)
|
||||
|
||||
@classmethod
|
||||
def get_requirements(cls):
|
||||
return [
|
||||
requirements.ModuleRequirement(
|
||||
name="kernel",
|
||||
description="Windows kernel",
|
||||
architectures=["Intel32", "Intel64"],
|
||||
),
|
||||
requirements.VersionRequirement(
|
||||
name="pslist", component=pslist.PsList, version=(3, 0, 0)
|
||||
),
|
||||
requirements.VersionRequirement(
|
||||
name="psscan", component=psscan.PsScan, version=(2, 0, 0)
|
||||
),
|
||||
requirements.VersionRequirement(
|
||||
name="thrdscan", component=thrdscan.ThrdScan, version=(2, 0, 0)
|
||||
),
|
||||
requirements.VersionRequirement(
|
||||
name="handles", component=handles.Handles, version=(4, 0, 0)
|
||||
),
|
||||
requirements.BooleanRequirement(
|
||||
name="physical-offsets",
|
||||
description="List processes with physical offsets instead of virtual offsets.",
|
||||
optional=True,
|
||||
),
|
||||
]
|
||||
|
||||
def _proc_name_to_string(self, proc):
|
||||
return proc.ImageFileName.cast(
|
||||
"string", max_length=proc.ImageFileName.vol.count, errors="replace"
|
||||
)
|
||||
|
||||
def _is_valid_proc_name(self, string: str) -> bool:
|
||||
return all(c in self.valid_proc_name_chars for c in string)
|
||||
|
||||
def _filter_garbage_procs(
|
||||
self, proc_list: Iterable[extensions.EPROCESS]
|
||||
) -> List[extensions.EPROCESS]:
|
||||
return [
|
||||
p
|
||||
for p in proc_list
|
||||
if p.is_valid() and self._is_valid_proc_name(self._proc_name_to_string(p))
|
||||
]
|
||||
|
||||
def _translate_offset(self, offset: int) -> int:
|
||||
if not self.config["physical-offsets"]:
|
||||
return offset
|
||||
|
||||
kernel = self.context.modules[self.config["kernel"]]
|
||||
layer_name = kernel.layer_name
|
||||
|
||||
try:
|
||||
_original_offset, _original_length, offset, _length, _layer_name = list(
|
||||
self.context.layers[layer_name].mapping(offset=offset, length=0)
|
||||
)[0]
|
||||
except exceptions.PagedInvalidAddressException:
|
||||
vollog.debug(f"Page fault: unable to translate {offset:0x}")
|
||||
|
||||
return offset
|
||||
|
||||
def _proc_list_to_dict(
|
||||
self, tasks: Iterable[extensions.EPROCESS]
|
||||
) -> Dict[int, extensions.EPROCESS]:
|
||||
tasks = self._filter_garbage_procs(tasks)
|
||||
return {self._translate_offset(proc.vol.offset): proc for proc in tasks}
|
||||
|
||||
def _check_pslist(self, tasks):
|
||||
return self._proc_list_to_dict(tasks)
|
||||
|
||||
def _check_psscan(
|
||||
self,
|
||||
) -> Dict[int, extensions.EPROCESS]:
|
||||
res = psscan.PsScan.scan_processes(
|
||||
context=self.context, kernel_module_name=self.config["kernel"]
|
||||
)
|
||||
|
||||
return self._proc_list_to_dict(res)
|
||||
|
||||
def _check_thrdscan(self) -> Dict[int, extensions.EPROCESS]:
|
||||
ret = []
|
||||
|
||||
for ethread in thrdscan.ThrdScan.scan_threads(
|
||||
self.context, module_name="kernel"
|
||||
):
|
||||
process = None
|
||||
try:
|
||||
process = ethread.owning_process()
|
||||
if not process.is_valid():
|
||||
continue
|
||||
|
||||
ret.append(process)
|
||||
except AttributeError:
|
||||
vollog.log(
|
||||
constants.LOGLEVEL_VVV,
|
||||
"Unable to find the owning process of ethread",
|
||||
)
|
||||
|
||||
return self._proc_list_to_dict(ret)
|
||||
|
||||
def _check_csrss_handles(
|
||||
self, tasks: Iterable[extensions.EPROCESS]
|
||||
) -> Dict[int, extensions.EPROCESS]:
|
||||
ret: List[extensions.EPROCESS] = []
|
||||
|
||||
type_map = handles.Handles.get_type_map(
|
||||
context=self.context, kernel_module_name=self.config["kernel"]
|
||||
)
|
||||
|
||||
cookie = handles.Handles.find_cookie(
|
||||
context=self.context, kernel_module_name=self.config["kernel"]
|
||||
)
|
||||
|
||||
for p in tasks:
|
||||
name = self._proc_name_to_string(p)
|
||||
if name != "csrss.exe":
|
||||
continue
|
||||
|
||||
try:
|
||||
ret += [
|
||||
handle.Body.cast("_EPROCESS")
|
||||
for handle in handles.Handles.handles(
|
||||
context=self.context,
|
||||
kernel_module_name=self.config["kernel"],
|
||||
handle_table=p.ObjectTable,
|
||||
)
|
||||
if handle.get_object_type(type_map, cookie) == "Process"
|
||||
]
|
||||
except exceptions.InvalidAddressException:
|
||||
vollog.log(
|
||||
constants.LOGLEVEL_VVV, "Cannot access eprocess object table"
|
||||
)
|
||||
|
||||
return self._proc_list_to_dict(ret)
|
||||
|
||||
def _generator(self):
|
||||
kdbg_list_processes = list(
|
||||
pslist.PsList.list_processes(
|
||||
context=self.context, kernel_module_name=self.config["kernel"]
|
||||
)
|
||||
)
|
||||
|
||||
# get processes from each source
|
||||
processes: Dict[str, Dict[int, extensions.EPROCESS]] = {}
|
||||
|
||||
processes["pslist"] = self._check_pslist(kdbg_list_processes)
|
||||
processes["psscan"] = self._check_psscan()
|
||||
processes["thrdscan"] = self._check_thrdscan()
|
||||
processes["csrss"] = self._check_csrss_handles(kdbg_list_processes)
|
||||
|
||||
# Unique set of all offsets from all sources
|
||||
offsets = set(chain(*(mapping.keys() for mapping in processes.values())))
|
||||
|
||||
for offset in offsets:
|
||||
# We know there will be at least one process mapped to each offset
|
||||
proc: extensions.EPROCESS = next(
|
||||
mapping[offset] for mapping in processes.values() if offset in mapping
|
||||
)
|
||||
|
||||
in_sources = {src: False for src in processes}
|
||||
|
||||
for source, process_mapping in processes.items():
|
||||
if offset in process_mapping:
|
||||
in_sources[source] = True
|
||||
|
||||
pid = proc.UniqueProcessId
|
||||
name = self._proc_name_to_string(proc)
|
||||
|
||||
exit_time = proc.get_exit_time()
|
||||
if type(exit_time) is not datetime.datetime:
|
||||
exit_time = ""
|
||||
else:
|
||||
exit_time = str(exit_time)
|
||||
|
||||
yield (
|
||||
0,
|
||||
(
|
||||
format_hints.Hex(offset),
|
||||
name,
|
||||
pid,
|
||||
in_sources["pslist"],
|
||||
in_sources["psscan"],
|
||||
in_sources["thrdscan"],
|
||||
in_sources["csrss"],
|
||||
exit_time,
|
||||
),
|
||||
)
|
||||
|
||||
def run(self):
|
||||
offset_type = "(Physical)" if self.config["physical-offsets"] else "(Virtual)"
|
||||
offset_str = "Offset" + offset_type
|
||||
|
||||
return renderers.TreeGrid(
|
||||
[
|
||||
(offset_str, format_hints.Hex),
|
||||
("Name", str),
|
||||
("PID", int),
|
||||
("pslist", bool),
|
||||
("psscan", bool),
|
||||
("thrdscan", bool),
|
||||
("csrss", bool),
|
||||
("Exit Time", str),
|
||||
],
|
||||
self._generator(),
|
||||
)
|
||||
@@ -1,24 +1,24 @@
|
||||
import datetime
|
||||
# This file is Copyright 2025 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
|
||||
import string
|
||||
from itertools import chain
|
||||
from typing import Dict, Iterable, List
|
||||
|
||||
from volatility3.framework import constants, exceptions, renderers
|
||||
from volatility3.framework.configuration import requirements
|
||||
from volatility3.framework.interfaces import plugins
|
||||
from volatility3.framework.renderers import format_hints
|
||||
from volatility3.framework.symbols.windows import extensions
|
||||
from volatility3.plugins.windows import handles, pslist, psscan, thrdscan
|
||||
from volatility3.framework import interfaces, deprecation
|
||||
from volatility3.plugins.windows.malware import psxview
|
||||
|
||||
vollog = logging.getLogger(__name__)
|
||||
|
||||
|
||||
class PsXView(plugins.PluginInterface):
|
||||
class PsXView(
|
||||
interfaces.plugins.PluginInterface,
|
||||
deprecation.PluginRenameClass,
|
||||
replacement_class=psxview.PsXView,
|
||||
removal_date="2026-06-07",
|
||||
):
|
||||
"""Lists all processes found via four of the methods described in \"The Art of Memory Forensics\" which may help \
|
||||
identify processes that are trying to hide themselves.
|
||||
identify processes that are trying to hide themselves.
|
||||
|
||||
We recommend using -r pretty if you are looking at this plugin's output in a terminal."""
|
||||
We recommend using -r pretty if you are looking at this plugin's output in a terminal.
|
||||
deprecated."""
|
||||
|
||||
# I've omitted the desktop thread scanning method because Volatility3 doesn't appear to have the functionality
|
||||
# which the original plugin used to do it.
|
||||
@@ -30,212 +30,3 @@ We recommend using -r pretty if you are looking at this plugin's output in a ter
|
||||
|
||||
_required_framework_version = (2, 0, 0)
|
||||
_version = (1, 0, 0)
|
||||
|
||||
valid_proc_name_chars = set(
|
||||
string.ascii_lowercase + string.ascii_uppercase + "." + " "
|
||||
)
|
||||
|
||||
@classmethod
|
||||
def get_requirements(cls):
|
||||
return [
|
||||
requirements.ModuleRequirement(
|
||||
name="kernel",
|
||||
description="Windows kernel",
|
||||
architectures=["Intel32", "Intel64"],
|
||||
),
|
||||
requirements.VersionRequirement(
|
||||
name="pslist", component=pslist.PsList, version=(3, 0, 0)
|
||||
),
|
||||
requirements.VersionRequirement(
|
||||
name="psscan", component=psscan.PsScan, version=(2, 0, 0)
|
||||
),
|
||||
requirements.VersionRequirement(
|
||||
name="thrdscan", component=thrdscan.ThrdScan, version=(2, 0, 0)
|
||||
),
|
||||
requirements.VersionRequirement(
|
||||
name="handles", component=handles.Handles, version=(4, 0, 0)
|
||||
),
|
||||
requirements.BooleanRequirement(
|
||||
name="physical-offsets",
|
||||
description="List processes with physical offsets instead of virtual offsets.",
|
||||
optional=True,
|
||||
),
|
||||
]
|
||||
|
||||
def _proc_name_to_string(self, proc):
|
||||
return proc.ImageFileName.cast(
|
||||
"string", max_length=proc.ImageFileName.vol.count, errors="replace"
|
||||
)
|
||||
|
||||
def _is_valid_proc_name(self, string: str) -> bool:
|
||||
return all(c in self.valid_proc_name_chars for c in string)
|
||||
|
||||
def _filter_garbage_procs(
|
||||
self, proc_list: Iterable[extensions.EPROCESS]
|
||||
) -> List[extensions.EPROCESS]:
|
||||
return [
|
||||
p
|
||||
for p in proc_list
|
||||
if p.is_valid() and self._is_valid_proc_name(self._proc_name_to_string(p))
|
||||
]
|
||||
|
||||
def _translate_offset(self, offset: int) -> int:
|
||||
if not self.config["physical-offsets"]:
|
||||
return offset
|
||||
|
||||
kernel = self.context.modules[self.config["kernel"]]
|
||||
layer_name = kernel.layer_name
|
||||
|
||||
try:
|
||||
_original_offset, _original_length, offset, _length, _layer_name = list(
|
||||
self.context.layers[layer_name].mapping(offset=offset, length=0)
|
||||
)[0]
|
||||
except exceptions.PagedInvalidAddressException:
|
||||
vollog.debug(f"Page fault: unable to translate {offset:0x}")
|
||||
|
||||
return offset
|
||||
|
||||
def _proc_list_to_dict(
|
||||
self, tasks: Iterable[extensions.EPROCESS]
|
||||
) -> Dict[int, extensions.EPROCESS]:
|
||||
tasks = self._filter_garbage_procs(tasks)
|
||||
return {self._translate_offset(proc.vol.offset): proc for proc in tasks}
|
||||
|
||||
def _check_pslist(self, tasks):
|
||||
return self._proc_list_to_dict(tasks)
|
||||
|
||||
def _check_psscan(
|
||||
self,
|
||||
) -> Dict[int, extensions.EPROCESS]:
|
||||
res = psscan.PsScan.scan_processes(
|
||||
context=self.context, kernel_module_name=self.config["kernel"]
|
||||
)
|
||||
|
||||
return self._proc_list_to_dict(res)
|
||||
|
||||
def _check_thrdscan(self) -> Dict[int, extensions.EPROCESS]:
|
||||
ret = []
|
||||
|
||||
for ethread in thrdscan.ThrdScan.scan_threads(
|
||||
self.context, module_name="kernel"
|
||||
):
|
||||
process = None
|
||||
try:
|
||||
process = ethread.owning_process()
|
||||
if not process.is_valid():
|
||||
continue
|
||||
|
||||
ret.append(process)
|
||||
except AttributeError:
|
||||
vollog.log(
|
||||
constants.LOGLEVEL_VVV,
|
||||
"Unable to find the owning process of ethread",
|
||||
)
|
||||
|
||||
return self._proc_list_to_dict(ret)
|
||||
|
||||
def _check_csrss_handles(
|
||||
self, tasks: Iterable[extensions.EPROCESS]
|
||||
) -> Dict[int, extensions.EPROCESS]:
|
||||
ret: List[extensions.EPROCESS] = []
|
||||
|
||||
type_map = handles.Handles.get_type_map(
|
||||
context=self.context, kernel_module_name=self.config["kernel"]
|
||||
)
|
||||
|
||||
cookie = handles.Handles.find_cookie(
|
||||
context=self.context, kernel_module_name=self.config["kernel"]
|
||||
)
|
||||
|
||||
for p in tasks:
|
||||
name = self._proc_name_to_string(p)
|
||||
if name != "csrss.exe":
|
||||
continue
|
||||
|
||||
try:
|
||||
ret += [
|
||||
handle.Body.cast("_EPROCESS")
|
||||
for handle in handles.Handles.handles(
|
||||
context=self.context,
|
||||
kernel_module_name=self.config["kernel"],
|
||||
handle_table=p.ObjectTable,
|
||||
)
|
||||
if handle.get_object_type(type_map, cookie) == "Process"
|
||||
]
|
||||
except exceptions.InvalidAddressException:
|
||||
vollog.log(
|
||||
constants.LOGLEVEL_VVV, "Cannot access eprocess object table"
|
||||
)
|
||||
|
||||
return self._proc_list_to_dict(ret)
|
||||
|
||||
def _generator(self):
|
||||
kdbg_list_processes = list(
|
||||
pslist.PsList.list_processes(
|
||||
context=self.context, kernel_module_name=self.config["kernel"]
|
||||
)
|
||||
)
|
||||
|
||||
# get processes from each source
|
||||
processes: Dict[str, Dict[int, extensions.EPROCESS]] = {}
|
||||
|
||||
processes["pslist"] = self._check_pslist(kdbg_list_processes)
|
||||
processes["psscan"] = self._check_psscan()
|
||||
processes["thrdscan"] = self._check_thrdscan()
|
||||
processes["csrss"] = self._check_csrss_handles(kdbg_list_processes)
|
||||
|
||||
# Unique set of all offsets from all sources
|
||||
offsets = set(chain(*(mapping.keys() for mapping in processes.values())))
|
||||
|
||||
for offset in offsets:
|
||||
# We know there will be at least one process mapped to each offset
|
||||
proc: extensions.EPROCESS = next(
|
||||
mapping[offset] for mapping in processes.values() if offset in mapping
|
||||
)
|
||||
|
||||
in_sources = {src: False for src in processes}
|
||||
|
||||
for source, process_mapping in processes.items():
|
||||
if offset in process_mapping:
|
||||
in_sources[source] = True
|
||||
|
||||
pid = proc.UniqueProcessId
|
||||
name = self._proc_name_to_string(proc)
|
||||
|
||||
exit_time = proc.get_exit_time()
|
||||
if type(exit_time) is not datetime.datetime:
|
||||
exit_time = ""
|
||||
else:
|
||||
exit_time = str(exit_time)
|
||||
|
||||
yield (
|
||||
0,
|
||||
(
|
||||
format_hints.Hex(offset),
|
||||
name,
|
||||
pid,
|
||||
in_sources["pslist"],
|
||||
in_sources["psscan"],
|
||||
in_sources["thrdscan"],
|
||||
in_sources["csrss"],
|
||||
exit_time,
|
||||
),
|
||||
)
|
||||
|
||||
def run(self):
|
||||
offset_type = "(Physical)" if self.config["physical-offsets"] else "(Virtual)"
|
||||
offset_str = "Offset" + offset_type
|
||||
|
||||
return renderers.TreeGrid(
|
||||
[
|
||||
(offset_str, format_hints.Hex),
|
||||
("Name", str),
|
||||
("PID", int),
|
||||
("pslist", bool),
|
||||
("psscan", bool),
|
||||
("thrdscan", bool),
|
||||
("csrss", bool),
|
||||
("Exit Time", str),
|
||||
],
|
||||
self._generator(),
|
||||
)
|
||||
|
||||
Reference in New Issue
Block a user