Merge remote-tracking branch 'upstream/develop' into linux_pslist_pstree_improvements

This commit is contained in:
Gustavo Moreira
2022-04-28 09:49:17 +10:00
74 changed files with 2256 additions and 338 deletions
+1 -4
View File
@@ -16,7 +16,4 @@ formats: all
python:
version: 3.7
install:
- method: pip
path: .
extra_requirements:
- doc
- requirements: doc/requirements.txt
+19
View File
@@ -4,6 +4,25 @@ API Changes
When an addition to the existing API is made, the minor version is bumped.
When an API feature or function is removed or changed, the major version is bumped.
2.1.0
=====
Add in the linux `task.get_threads` method added to the API.
2.0.3
=====
`DEVICE_OBJECT.get_attached_devices` and `DRIVER_OBJECT.get_devices` added to the API.
2.0.2
=====
Fix the behaviour of the offsets returned by the PDB scanner.
2.0.0
=====
Remove the `symbol_shift` mechanism, where symbol tables could alter their own symbols.
Symbols from a symbol table are now always the offset values. They can be added to a Module
and when symbols are requested from a Module they are shifted by the module's offset to get
an absolute offset. This can be done with `Module.get_absolute_symbol_address` or as part of
`Module.object_from_symbol(absolute = False, ...)`.
1.2.0
=====
+1 -1
View File
@@ -1,6 +1,6 @@
prune development
include * .*
include doc/make.bat doc/Makefile
include doc/make.bat doc/Makefile doc/requirements.txt
recursive-include doc/source *
recursive-include volatility3 *.json
recursive-exclude doc/source volatility3.*.rst
+5 -3
View File
@@ -14,7 +14,9 @@ technical and performance challenges associated with the original
code base that became apparent over the previous 10 years. Another benefit
of the rewrite is that Volatility 3 could be released under a custom
license that was more aligned with the goals of the Volatility community,
the Volatility Software License (VSL). See the [LICENSE](LICENSE.txt) file for more details.
the Volatility Software License (VSL). See the
[LICENSE](https://www.volatilityfoundation.org/license/vsl-v1.0) file for
more details.
## Requirements
@@ -39,7 +41,7 @@ pip3 install -r requirements.txt
## Downloading Volatility
The latest stable version of Volatility will always be the master branch of the GitHub repository. You can get the latest version of the code using the following command:
The latest stable version of Volatility will always be the stable branch of the GitHub repository. You can get the latest version of the code using the following command:
```shell
git clone https://github.com/volatilityfoundation/volatility3.git
@@ -102,7 +104,7 @@ The latest generated copy of the documentation can be found at: <https://volatil
## Licensing and Copyright
Copyright (C) 2007-2021 Volatility Foundation
Copyright (C) 2007-2022 Volatility Foundation
All Rights Reserved
+2 -2
View File
@@ -1,4 +1,4 @@
# These packages are required for building the documentation.
sphinx>=1.8.2
sphinx>=4.0.0
sphinx_autodoc_typehints>=1.4.0
sphinx-rtd-theme>=0.4.3
sphinx-rtd-theme>=0.4.3
+1 -1
View File
@@ -300,7 +300,7 @@ This will mean that when a specific structure is loaded from the symbol_space, i
`StructType`, but instead is instantiated using the NewStructureClass, meaning new methods can be called directly on it.
If the situation really calls for an entirely new object, that isn't covered by one of the existing
:py:class:`~volatility3.framework.objects.PrimativeObject` objects (such as
:py:class:`~volatility3.framework.objects.PrimitiveObject` objects (such as
:py:class:`~volatility3.framework.objects.Integer`,
:py:class:`~volatility3.framework.objects.Boolean`,
:py:class:`~volatility3.framework.objects.Float`,
+11 -2
View File
@@ -1,4 +1,4 @@
# This file is Copyright 2019 Volatility Foundation and licensed under the Volatility Software License 1.0
# This file is Copyright 2022 Volatility Foundation and licensed under the Volatility Software License 1.0
# which is available at https://www.volatilityfoundation.org/license/vsl-v1.0
#
#
@@ -84,6 +84,15 @@ def setup(app):
for line in submodule_lines:
contents.write(line.replace(b'volatility3.framework.plugins', b'volatility3.plugins'))
# Clear up the framework.plugins page
with open(os.path.join(os.path.dirname(__file__), 'volatility3.framework.plugins.rst'), "rb") as contents:
real_lines = contents.readlines()
with open(os.path.join(os.path.dirname(__file__), 'volatility3.framework.plugins.rst'), "wb") as contents:
for line in real_lines:
if b'volatility3.framework.plugins.' not in line:
contents.write(line)
# If extensions (or modules to document with autodoc) are in another directory,
# add these directories to sys.path here. If the directory is relative to the
@@ -126,7 +135,7 @@ master_doc = 'index'
# General information about the project.
project = 'Volatility 3'
copyright = '2012-2019, Volatility Foundation'
copyright = '2012-2022, Volatility Foundation'
# The version info for the project you're documenting, acts as replacement for
# |version| and |release|, also used in various other places throughout the
+8
View File
@@ -0,0 +1,8 @@
Writing Plugins
===============
.. toctree::
simple-plugin
complex-plugin
using-as-a-library
+1 -1
View File
@@ -145,7 +145,7 @@ Struct, Structure
Symbol
This is used in many different contexts, as a short term for many things. Within Volatility, a symbol is a
construct that usually encompasses a specific type :ref:`type<Type>` at a specfific :ref:`offset<Offset>`,
construct that usually encompasses a specific type :ref:`type<Type>` at a specific :ref:`offset<Offset>`,
representing a particular instance of that type within the memory of a compiled and running program. An example
would be the location in memory of a list of active tcp endpoints maintained by the networking stack
within an operating system.
+2 -4
View File
@@ -12,11 +12,9 @@ Here are some guidelines for using Volatility 3 effectively:
.. toctree::
basics
simple-plugin
vol2to3
complex-plugin
using-as-a-library
development
symbol-tables
vol2to3
volshell
glossary
+3 -3
View File
@@ -196,7 +196,7 @@ The plugin then defaults the ``BaseDllName`` and ``FullDllName`` variables to an
which is a way of indicating to the user interface that the value couldn't be read for some reason (but that it isn't fatal).
There are currently four different reasons a value may be unreadable:
* **Unreadble**: values which are empty because the data cannot be read
* **Unreadable**: values which are empty because the data cannot be read
* **Unparsable**: values which are empty because the data cannot be interpreted correctly
* **NotApplicable**: values which are empty because they don't make sense for this particular entry
* **NotAvailable**: values which cannot be provided now (but might in a future run, via new symbols or an updated plugin)
@@ -206,9 +206,9 @@ information may not be provided.
The plugin then takes the process's ``BaseDllName`` value, and calls :py:meth:`~volatility3.framework.symbols.windows.extensions.UNICODE_STRING.get_string` on it. All structure attributes,
as defined by the symbols, are directly accessible and use the case-style of the symbol library it came from (in Windows,
attributes are CamelCase), such as ``entry.BaseDllName`` in this instance. Any attribtues not defined by the symbol but added
attributes are CamelCase), such as ``entry.BaseDllName`` in this instance. Any attributes not defined by the symbol but added
by Volatility extensions cannot be properties (in case they overlap with the attributes defined in the symbol libraries)
and are therefore always methods and prepended with ``get_``, in this example ``BaseDllName.get_string()``.
and are therefore always methods and pretended with ``get_``, in this example ``BaseDllName.get_string()``.
Finally, ``FullDllName`` is populated. These operations read from memory, and as such, the memory image may be unable to
read the data at a particular offset. This will cause an exception to be thrown. In Volatility 3, exceptions are thrown
+18
View File
@@ -76,3 +76,21 @@ The banners available for volatility to use can be found using the `isfinfo` plu
long time to run depending on the number of JSON files available. This will list all the JSON (ISF) files that
volatility3 is aware of, and for linux/mac systems what banner string they search for. For volatility to use the JSON
file, the banners must match exactly (down to the compilation date).
.. note::
Steps for constructing a new kernel ISF JSON file:
* Run the `banners` plugin on the image to determine the necessary kernel
* Locate a copy of the debug kernel that matches the identified banner
* Clone or update the dwarf2json repo: :code:`git clone https://github.com/volatilityfoundation/dwarf2json`
* Run :code:`go build` in the directory if the source has changed
* Run :code:`dwarf2json linux --elf [path to debug kernel] > [kernel name].json`
* For Mac change `linux` to `mac`
* Copy the `.json` file to the symbols directory into `[symbols directory]/linux`
* For Mac change `linux` to `mac`
+2 -1
View File
@@ -131,7 +131,8 @@ A suitable list of automagics for a particular plugin (based on operating system
automagics = automagic.choose_automagic(available_automagics, plugin)
This will take the plugin module, extract the operating system (first level of the hierarchy) and then return just
the automagics which apply to the operating system.
the automagics which apply to the operating system. Each automagic can exclude itself from being used for specific
operating systems, so that an automagic designed for linux is not used for windows or mac plugins.
These automagics can then be run by providing the list, the context, the plugin to be run, the hierarchy name that
the plugin will be constructed on ('plugins' by default) and a progress_callback. This is a callable which takes
+4
View File
@@ -62,6 +62,10 @@ automagic processes are clearly defined and can be enabled or disabled as necess
included a stacker automagic to emulate the most common feature of Volatility 2, automatically stacking address spaces
(now translation layers) on top of each other.
By default the automagic chosen to be run are determined based on the plugin requested, so that linux plugins get linux
specific automagic and windows plugins get windows specific automagic. This should reduce unnecessarily searching for
linux kernels in a windows image, for example. At the moment this is not user configurableS.
Searching and Scanning
----------------------
Scanning is very similar to scanning in Volatility 2, a scanner object (such as a
+45 -42
View File
@@ -22,15 +22,17 @@ be run.
When volshell starts, it will show the version of volshell, a brief message indicating how to get more help, the current
operating system mode for volshell, and the current layer available for use.
.. code-block:: python
::
Volshell (Volatility 3 Framework) 1.0.1
Volshell (Volatility 3 Framework) 2.0.2
Readline imported successfully PDB scanning finished
Call help() to see available functions
Volshell mode: Generic
Current Layer: primary
Volshell mode : Generic
Current Layer : primary
Current Symbol Table : None
Current Kernel Name : None
(primary) >>>
@@ -53,11 +55,11 @@ run our examples against.
We'll start by creating a process variable, and putting the first result from `ps()` in it. Since the shell is a
python environment, we can do the following:
.. code-block:: python
::
(primary) >>> proc = ps()[0]
(primary) >>> proc
<EPROCESS nt_symbols1!_EPROCESS: primary @ 0x8c0bcac87040 #2624>
(layer_name) >>> proc = ps()[0]
(layer_name) >>> proc
<EPROCESS symbol_table_name1!_EPROCESS: layer_name @ 0xe08ff2459040 #1968>
When printing a volatility structure, various information is output, in this case the `type_name`, the `layer` and
`offset` that it's been constructed on, and the size of the structure.
@@ -68,41 +70,41 @@ built-in mechanism for providing more information about a structure, called `dis
either a type name (which if not prefixed with symbol table name, will use the kernel symbol table identified by the
automagic).
.. code-block:: python
::
(primary) >>> dt('_EPROCESS')
nt_symbols1!_EPROCESS (2624 bytes)
0x0 : Pcb nt_symbols1!_KPROCESS
0x438 : ProcessLock nt_symbols1!_EX_PUSH_LOCK
0x440 : UniqueProcessId nt_symbols1!pointer
0x448 : ActiveProcessLinks nt_symbols1!_LIST_ENTRY
(layer_name) >>> dt('_EPROCESS')
symbol_table_name1!_EPROCESS (1968 bytes)
0x0 : Pcb symbol_table_name1!_KPROCESS
0x2d8 : ProcessLock symbol_table_name1!_EX_PUSH_LOCK
0x2e0 : RundownProtect symbol_table_name1!_EX_RUNDOWN_REF
0x2e8 : UniqueProcessId symbol_table_name1!pointer
...
It can also be provided with an object and will interpret the data for each in the process:
.. code-block:: python
::
(primary) >>> dt(proc)
nt_symbols1!_EPROCESS (2624 bytes)
0x0 : Pcb nt_symbols1!_KPROCESS 0x8c0bccf8d040
0x438 : ProcessLock nt_symbols1!_EX_PUSH_LOCK 0x8c0bccf8d478
0x440 : UniqueProcessId nt_symbols1!pointer 356
0x448 : ActiveProcessLinks nt_symbols1!_LIST_ENTRY 0x8c0bccf8d488
(layer_name) >>> dt(proc)
symbol_table_name1!_EPROCESS (1968 bytes)
0x0 : Pcb symbol_table_name1!_KPROCESS 0xe08ff2459040
0x2d8 : ProcessLock symbol_table_name1!_EX_PUSH_LOCK 0xe08ff2459318
0x2e0 : RundownProtect symbol_table_name1!_EX_RUNDOWN_REF 0xe08ff2459320
0x2e8 : UniqueProcessId symbol_table_name1!pointer 4
...
These values can be accessed directory as attributes
.. code-block:: python
::
(primary) >>> proc.UniqueProcessId
(layer_name) >>> proc.UniqueProcessId
356
Pointer structures contain the value they point to, but attributes accessed are forwarded to the object they point to.
This means that pointers do not need to be explicitly dereferenced to access underling objects.
.. code-block:: python
::
(primary) >>> proc.Pcb.DirectoryTableBase
(layer_name) >>> proc.Pcb.DirectoryTableBase
4355817472
Running plugins
@@ -112,28 +114,28 @@ It's possible to run any plugin by importing it appropriately and passing it to
method. In the following example we'll provide no additional parameters. Volatility will show us which parameters
were required:
.. code-block:: python
::
(primary) >>> from volatility3.plugins.windows import pslist
(primary) >>> display_plugin_output(pslist.PsList)
Unable to validate the plugin requirements: ['plugins.Volshell.9QZLXJKFWESI0BAP3M1U7Y5VCT468GRN.PsList.primary', 'plugins.Volshell.9QZLXJKFWESI0BAP3M1U7Y5VCT468GRN.PsList.nt_symbols']
(layer_name) >>> from volatility3.plugins.windows import pslist
(layer_name) >>> display_plugin_output(pslist.PsList)
Unable to validate the plugin requirements: ['plugins.Volshell.VH3FSA1JBG0QP9E62Z8OT5UCIMLNYKW4.PsList.kernel']
We can see that it's made a temporary configuration path for the plugin, and that neither `primary` nor `nt_symbols`
was fulfilled.
We can see that it's made a temporary configuration path for the plugin, and that the `kernel` requirement
was not fulfilled.
We can see all the options that the plugin can accept by access the `get_requirements()` method of the plugin.
This is a classmethod, so can be called on an uninstantiated copy of the plugin.
.. code-block:: python
::
(primary) >>> pslist.PsList.get_requirements()
[<TranslationLayerRequirement: primary>, <SymbolTableRequirement: nt_symbols>, <BooleanRequirement: physical>, <ListRequirement: pid>, <BooleanRequirement: dump>]
(layer_name) >>> pslist.PsList.get_requirements()
[<ModuleRequirement: kernel>, <BooleanRequirement: physical>, <ListRequirement: pid>, <BooleanRequirement: dump>]
We can provide arguments via the `dpo` method call:
.. code-block:: python
::
(primary) >>> display_plugin_output(pslist.PsList, primary = self.current_layer, nt_symbols = self.config['nt_symbols'])
(layer_name) >>> display_plugin_output(pslist.PsList, kernel = self.config['kernel'])
PID PPID ImageFileName Offset(V) Threads Handles SessionId Wow64 CreateTime ExitTime File output
@@ -142,17 +144,18 @@ We can provide arguments via the `dpo` method call:
356 4 smss.exe 0x8c0bccf8d040 3 - N/A False 2021-03-13 17:25:33.000000 N/A Disabled
...
Here's we've provided the current layer as the TranslationLayerRequirement, and used the symbol tables requirement
requested by the volshell plugin itself. A different table could be loaded and provided instead. The context used
Here's we've provided the kernel name that was requested by the volshell plugin itself (the generic volshell does not
load a kernel module, and instead only has a TranslationLayerRequirement).
A different module could be created and provided instead. The context used
by the `dpo` method is always `context`.
Instead of print the results directly to screen, they can be gathered into a TreeGrid objects for direct access by
using the `generate_treegrid` or `gt` command.
.. code-block:: python
::
(primary) >>> treegrid = gt(pslist.PsList, primary = self.current_layer, nt_symbols = self.config['nt_symbols'])
(primary) >>> treegrid.populate()
(layer_name) >>> treegrid = gt(pslist.PsList, kernel = self.config['kernel'])
(layer_name) >>> treegrid.populate()
Treegrids must be populated before the data in them can be accessed. This is where the plugin actually runs and
produces data.
+20 -10
View File
@@ -19,14 +19,14 @@ import os
import sys
import tempfile
import traceback
from typing import Dict, Type, Union, Any
from typing import Any, Dict, Type, Union
from urllib import parse, request
import volatility3.plugins
import volatility3.symbols
from volatility3 import framework
from volatility3.cli import text_renderer, volargparse
from volatility3.framework import automagic, constants, contexts, exceptions, interfaces, plugins, configuration
from volatility3.framework import automagic, configuration, constants, contexts, exceptions, interfaces, plugins
from volatility3.framework.automagic import stacker
from volatility3.framework.configuration import requirements
@@ -157,6 +157,10 @@ class CommandLine:
help = "Write configuration JSON file out to config.json",
default = False,
action = 'store_true')
parser.add_argument("--save-config",
help = "Save configuration JSON file to a file",
default = None,
type = str)
parser.add_argument("--clear-cache",
help = "Clears out all short-term cached items",
default = False,
@@ -320,8 +324,13 @@ class CommandLine:
self.file_handler_class_factory())
if args.write_config:
vollog.debug("Writing out configuration data to config.json")
with open("config.json", "w") as f:
vollog.warning('Use of --write-config has been deprecated, replaced by --save-config <filename>')
args.save_config = 'config.json'
if args.save_config:
vollog.debug("Writing out configuration data to {args.save_config}")
if os.path.exists(os.path.abspath(args.save_config)):
parser.error(f"Cannot write configuration: file {args.save_config} already exists")
with open(args.save_config, "w") as f:
json.dump(dict(constructed.build_configuration()), f, sort_keys = True, indent = 2)
except exceptions.UnsatisfiedException as excp:
self.process_unsatisfied_exceptions(excp)
@@ -443,16 +452,17 @@ class CommandLine:
print(f"Unsatisfied requirement {config_path}: {excp.unsatisfied[config_path].description}")
if symbols_failed:
print("\nA symbol table requirement was not fulfilled. Please verify that:\n"
"\tYou have the correct symbol file for the requirement\n"
"\tThe symbol file is under the correct directory or zip file\n"
"\tThe symbol file is named appropriately or contains the correct banner\n")
if translation_failed:
print("\nA translation layer requirement was not fulfilled. Please verify that:\n"
"\tA file was provided to create this layer (by -f, --single-location or by config)\n"
"\tThe file exists and is readable\n"
"\tThe necessary symbols are present and identified by volatility3")
"\tThe file is a valid memory image and was acquired cleanly")
if symbols_failed:
print("\nA symbol table requirement was not fulfilled. Please verify that:\n"
"\tThe associated translation layer requirement was fulfilled\n"
"\tYou have the correct symbol file for the requirement\n"
"\tThe symbol file is under the correct directory or zip file\n"
"\tThe symbol file is named appropriately or contains the correct banner\n")
def populate_config(self, context: interfaces.context.ContextInterface,
configurables_list: Dict[str, Type[interfaces.configuration.ConfigurableInterface]],
+57 -26
View File
@@ -1,6 +1,7 @@
# 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 csv
import datetime
import json
import logging
@@ -8,7 +9,7 @@ import random
import string
import sys
from functools import wraps
from typing import Callable, Any, List, Tuple, Dict
from typing import Any, Callable, Dict, List, Tuple
from volatility3.framework import interfaces, renderers
from volatility3.framework.renderers import format_hints
@@ -66,7 +67,6 @@ def multitypedata_as_text(value: format_hints.MultiTypeData) -> str:
def optional(func: Callable) -> Callable:
@wraps(func)
def wrapped(x: Any) -> str:
if isinstance(x, interfaces.renderers.BaseAbsentValue):
@@ -80,7 +80,6 @@ def optional(func: Callable) -> Callable:
def quoted_optional(func: Callable) -> Callable:
@wraps(func)
def wrapped(x: Any) -> str:
result = optional(func)(x)
@@ -102,7 +101,7 @@ def display_disassembly(disasm: interfaces.renderers.Disassembly) -> str:
disasm: Input disassembly objects
Returns:
A string as rendererd by capstone where available, otherwise output as if it were just bytes
A string as rendered by capstone where available, otherwise output as if it were just bytes
"""
if CAPSTONE_PRESENT:
@@ -182,16 +181,28 @@ class QuickTextRenderer(CLIRenderer):
outfd.write("\n")
class NoneRenderer(CLIRenderer):
"""Outputs no results"""
name = "none"
def get_render_options(self):
pass
def render(self, grid: interfaces.renderers.TreeGrid) -> None:
if not grid.populated:
grid.populate(lambda x, y: True, True)
class CSVRenderer(CLIRenderer):
_type_renderers = {
format_hints.Bin: quoted_optional(lambda x: f"0b{x:b}"),
format_hints.Hex: quoted_optional(lambda x: f"0x{x:x}"),
format_hints.HexBytes: quoted_optional(hex_bytes_as_text),
format_hints.MultiTypeData: quoted_optional(multitypedata_as_text),
interfaces.renderers.Disassembly: quoted_optional(display_disassembly),
bytes: quoted_optional(lambda x: " ".join([f"{b:02x}" for b in x])),
datetime.datetime: quoted_optional(lambda x: x.strftime("%Y-%m-%d %H:%M:%S.%f %Z")),
'default': quoted_optional(lambda x: f"{x}")
format_hints.Bin: optional(lambda x: f"0b{x:b}"),
format_hints.Hex: optional(lambda x: f"0x{x:x}"),
format_hints.HexBytes: optional(hex_bytes_as_text),
format_hints.MultiTypeData: optional(multitypedata_as_text),
interfaces.renderers.Disassembly: optional(display_disassembly),
bytes: optional(lambda x: " ".join([f"{b:02x}" for b in x])),
datetime.datetime: optional(lambda x: x.strftime("%Y-%m-%d %H:%M:%S.%f %Z")),
'default': optional(lambda x: f"{x}")
}
name = "csv"
@@ -208,28 +219,28 @@ class CSVRenderer(CLIRenderer):
"""
outfd = sys.stdout
line = ['"TreeDepth"']
header_list = ['TreeDepth']
for column in grid.columns:
# Ignore the type because namedtuples don't realize they have accessible attributes
line.append("{}".format('"' + column.name + '"'))
outfd.write(f"{','.join(line)}")
header_list.append(f"{column.name}")
writer = csv.DictWriter(outfd, header_list)
writer.writeheader()
def visitor(node: interfaces.renderers.TreeNode, accumulator):
accumulator.write("\n")
# Nodes always have a path value, giving them a path_depth of at least 1, we use max just in case
accumulator.write(str(max(0, node.path_depth - 1)) + ",")
line = []
row = {'TreeDepth': str(max(0, node.path_depth - 1))}
for column_index in range(len(grid.columns)):
column = grid.columns[column_index]
renderer = self._type_renderers.get(column.type, self._type_renderers['default'])
line.append(renderer(node.values[column_index]))
accumulator.write(f"{','.join(line)}")
row[f'{column.name}'] = renderer(node.values[column_index])
accumulator.writerow(row)
return accumulator
if not grid.populated:
grid.populate(visitor, outfd)
grid.populate(visitor, writer)
else:
grid.visit(node = None, function = visitor, initial_accumulator = outfd)
grid.visit(node = None, function = visitor, initial_accumulator = writer)
outfd.write("\n")
@@ -263,7 +274,8 @@ class PrettyTextRenderer(CLIRenderer):
max_column_widths = dict([(column.name, len(column.name)) for column in grid.columns])
def visitor(
node: interfaces.renderers.TreeNode, accumulator: List[Tuple[int, Dict[interfaces.renderers.Column, bytes]]]
node: interfaces.renderers.TreeNode,
accumulator: List[Tuple[int, Dict[interfaces.renderers.Column, bytes]]]
) -> List[Tuple[int, Dict[interfaces.renderers.Column, bytes]]]:
# Nodes always have a path value, giving them a path_depth of at least 1, we use max just in case
max_column_widths[tree_indent_column] = max(max_column_widths.get(tree_indent_column, 0), node.path_depth)
@@ -272,9 +284,10 @@ class PrettyTextRenderer(CLIRenderer):
column = grid.columns[column_index]
renderer = self._type_renderers.get(column.type, self._type_renderers['default'])
data = renderer(node.values[column_index])
field_width = max([len(self.tab_stop(x)) for x in f"{data}".split("\n")])
max_column_widths[column.name] = max(max_column_widths.get(column.name, len(column.name)),
len(f"{data}"))
line[column] = data
field_width)
line[column] = data.split("\n")
accumulator.append((node.path_depth, line))
return accumulator
@@ -296,7 +309,25 @@ class PrettyTextRenderer(CLIRenderer):
column_titles = [""] + [column.name for column in grid.columns]
outfd.write(format_string.format(*column_titles))
for (depth, line) in final_output:
outfd.write(format_string.format("*" * depth, *[line[column] for column in grid.columns]))
nums_line = max([len(line[column]) for column in line])
for column in line:
line[column] = line[column] + ([""] * (nums_line - len(line[column])))
for index in range(nums_line):
if index == 0:
outfd.write(format_string.format("*" * depth, *[self.tab_stop(line[column][index]) for column in grid.columns]))
else:
outfd.write(format_string.format(" " * depth, *[self.tab_stop(line[column][index]) for column in grid.columns]))
def tab_stop(self, line: str) -> str:
tab_width = 8
while line.find('\t') >= 0:
i = line.find('\t')
if (tab_width > 0):
pad = " " * (tab_width - (i % tab_width))
else:
pad = ""
line = line.replace("\t", pad, 1)
return line
class JsonRenderer(CLIRenderer):
+19 -7
View File
@@ -7,12 +7,11 @@ import json
import logging
import os
import sys
import glob
import volatility3.plugins
import volatility3.symbols
from volatility3 import cli, framework
from volatility3.cli.volshell import generic, windows, linux, mac
from volatility3.cli.volshell import generic, linux, mac, windows
from volatility3.framework import automagic, constants, contexts, exceptions, interfaces, plugins
# Make sure we log everything
@@ -86,6 +85,10 @@ class VolShell(cli.CommandLine):
help = "Write configuration JSON file out to config.json",
default = False,
action = 'store_true')
parser.add_argument("--save-config",
help = "Save configuration JSON file to a file",
default = None,
type = str)
parser.add_argument("--clear-cache",
help = "Clears out all short-term cached items",
default = False,
@@ -138,8 +141,7 @@ class VolShell(cli.CommandLine):
console.setLevel(10 - (partial_args.verbosity - 2))
if partial_args.clear_cache:
for cache_filename in glob.glob(os.path.join(constants.CACHE_PATH, '*.cache')):
os.unlink(cache_filename)
framework.clear_cache()
# Do the initialization
ctx = contexts.Context() # Construct a blank context
@@ -236,12 +238,22 @@ class VolShell(cli.CommandLine):
self.file_handler_class_factory())
if args.write_config:
vollog.debug("Writing out configuration data to config.json")
with open("config.json", "w") as f:
vollog.warning('Use of --write-config has been deprecated, replaced by --save-config <filename>')
args.save_config = 'config.json'
if args.save_config:
vollog.debug("Writing out configuration data to {args.save_config}")
if os.path.exists(os.path.abspath(args.save_config)):
parser.error(f"Cannot write configuration: file {args.save_config} already exists")
with open(args.save_config, "w") as f:
json.dump(dict(constructed.build_configuration()), f, sort_keys = True, indent = 2)
except exceptions.UnsatisfiedException as excp:
self.process_unsatisfied_exceptions(excp)
parser.exit(1, f"Unable to validate the plugin requirements: {[x for x in excp.unsatisfied]}\n")
try:
# Construct and run the plugin
constructed.run()
if constructed:
constructed.run()
except exceptions.VolatilityException as excp:
self.process_exceptions(excp)
parser.exit(1, f"Unable to validate the plugin requirements: {[x for x in excp.unsatisfied]}\n")
+60 -12
View File
@@ -8,11 +8,11 @@ import random
import string
import struct
import sys
from typing import Any, Dict, List, Optional, Tuple, Union, Type, Iterable
from urllib import request, parse
from typing import Any, Dict, Iterable, List, Optional, Tuple, Type, Union
from urllib import parse, request
from volatility3.cli import text_renderer, volshell
from volatility3.framework import renderers, interfaces, objects, plugins, exceptions
from volatility3.framework import exceptions, interfaces, objects, plugins, renderers
from volatility3.framework.configuration import requirements
from volatility3.framework.layers import intel, physical, resources
@@ -31,6 +31,8 @@ class Volshell(interfaces.plugins.PluginInterface):
def __init__(self, *args, **kwargs):
super().__init__(*args, **kwargs)
self.__current_layer: Optional[str] = None
self.__current_symbol_table: Optional[str] = None
self.__current_kernel_name: Optional[str] = None
self.__console = None
def random_string(self, length: int = 32) -> str:
@@ -57,8 +59,6 @@ class Volshell(interfaces.plugins.PluginInterface):
Return a TreeGrid but this is always empty since the point of this plugin is to run interactively
"""
self.__current_layer = self.config['primary']
# Try to enable tab completion
try:
import readline
@@ -79,9 +79,11 @@ class Volshell(interfaces.plugins.PluginInterface):
banner = f"""
Call help() to see available functions
Volshell mode: {mode}
Current Layer: {self.current_layer}
"""
Volshell mode : {mode}
Current Layer : {self.current_layer}
Current Symbol Table : {self.current_symbol_table}
Current Kernel Name : {self.current_kernel_name}
"""
sys.ps1 = f"({self.current_layer}) >>> "
self.__console = code.InteractiveConsole(locals = self._construct_locals_dict())
@@ -121,7 +123,10 @@ class Volshell(interfaces.plugins.PluginInterface):
(['dw', 'display_words'], self.display_words), (['dd',
'display_doublewords'], self.display_doublewords),
(['dq', 'display_quadwords'], self.display_quadwords), (['dis', 'disassemble'], self.disassemble),
(['cl', 'change_layer'], self.change_layer), (['context'], self.context), (['self'], self),
(['cl', 'change_layer'], self.change_layer),
(['cs', 'change_symboltable'], self.change_symbol_table),
(['ck', 'change_kernel'], self.change_kernel),
(['context'], self.context), (['self'], self),
(['dpo', 'display_plugin_output'], self.display_plugin_output),
(['gt', 'generate_treegrid'], self.generate_treegrid), (['rt',
'render_treegrid'], self.render_treegrid),
@@ -174,15 +179,58 @@ class Volshell(interfaces.plugins.PluginInterface):
@property
def current_layer(self):
if self.__current_layer is None:
self.__current_layer = self.config['primary']
return self.__current_layer
def change_layer(self, layer_name = None):
@property
def current_symbol_table(self):
if self.__current_symbol_table is None and self.kernel:
self.__current_symbol_table = self.kernel.symbol_table_name
return self.__current_symbol_table
@property
def current_kernel_name(self):
if self.__current_kernel_name is None:
self.__current_kernel_name = self.config.get('kernel', None)
return self.__current_kernel_name
@property
def kernel(self):
"""Returns the current kernel object"""
if self.current_kernel_name not in self.context.modules:
return None
return self.context.modules[self.current_kernel_name]
def change_layer(self, layer_name: str = None):
"""Changes the current default layer"""
if not layer_name:
layer_name = self.config['primary']
self.__current_layer = layer_name
layer_name = self.current_layer
if layer_name not in self.context.layers:
print(f"Layer {layer_name} not present in context")
else:
self.__current_layer = layer_name
sys.ps1 = f"({self.current_layer}) >>> "
def change_symbol_table(self, symbol_table_name: str = None):
"""Changes the current_symbol_table"""
if not symbol_table_name:
print("No symbol table provided, not changing current symbol table")
if symbol_table_name not in self.context.symbol_space:
print(f"Symbol table {symbol_table_name} not present in context symbol_space")
else:
self.__current_symbol_table = symbol_table_name
print(f"Current Symbol Table: {self.current_symbol_table}")
def change_kernel(self, kernel_name: str = None):
if not kernel_name:
print("No kernel module name provided, not changing current kernel")
if kernel_name not in self.context.modules:
print(f"Kernel module {kernel_name} not found in the context module list")
else:
self.__current_kernel_name = kernel_name
print(f"Current kernel : {self.current_kernel_name}")
def display_bytes(self, offset, count = 128, layer_name = None):
"""Displays byte values and ASCII characters"""
remaining_data = self._read_data(offset, count = count, layer_name = layer_name)
+12 -6
View File
@@ -5,7 +5,7 @@
from typing import Any, List, Tuple, Union
from volatility3.cli.volshell import generic
from volatility3.framework import interfaces, constants
from volatility3.framework import constants, interfaces
from volatility3.framework.configuration import requirements
from volatility3.plugins.linux import pslist
@@ -15,9 +15,9 @@ class Volshell(generic.Volshell):
@classmethod
def get_requirements(cls):
return (super().get_requirements() + [
requirements.SymbolTableRequirement(name = "vmlinux", description = "Linux kernel symbols"),
requirements.PluginRequirement(name = 'pslist', plugin = pslist.PsList, version = (1, 0, 0)),
return ([
requirements.ModuleRequirement(name = "kernel", description = "Linux kernel module"),
requirements.PluginRequirement(name = 'pslist', plugin = pslist.PsList, version = (2, 0, 0)),
requirements.IntRequirement(name = 'pid', description = "Process ID", optional = True)
])
@@ -37,14 +37,14 @@ class Volshell(generic.Volshell):
def list_tasks(self):
"""Returns a list of task objects from the primary layer"""
# We always use the main kernel memory and associated symbols
return list(pslist.PsList.list_tasks(self.context, self.config['primary'], self.config['vmlinux']))
return list(pslist.PsList.list_tasks(self.context, self.current_kernel_name))
def construct_locals(self) -> List[Tuple[List[str], Any]]:
result = super().construct_locals()
result += [
(['ct', 'change_task', 'cp'], self.change_task),
(['lt', 'list_tasks', 'ps'], self.list_tasks),
(['symbols'], self.context.symbol_space[self.config['vmlinux']]),
(['symbols'], self.context.symbol_space[self.current_symbol_table]),
]
if self.config.get('pid', None) is not None:
self.change_task(self.config['pid'])
@@ -64,3 +64,9 @@ class Volshell(generic.Volshell):
if symbol_table is None:
symbol_table = self.config['vmlinux']
return super().display_symbols(symbol_table)
@property
def current_layer(self):
if self.__current_layer is None:
self.__current_layer = self.kernel.layer_name
return self.__current_layer
+13 -7
View File
@@ -15,9 +15,9 @@ class Volshell(generic.Volshell):
@classmethod
def get_requirements(cls):
return (super().get_requirements() + [
requirements.SymbolTableRequirement(name = "darwin", description = "Darwin kernel symbols"),
requirements.PluginRequirement(name = 'pslist', plugin = pslist.PsList, version = (1, 0, 0)),
return ([
requirements.ModuleRequirement(name = "kernel", description = "Darwin kernel module"),
requirements.PluginRequirement(name = 'pslist', plugin = pslist.PsList, version = (3, 0, 0)),
requirements.IntRequirement(name = 'pid', description = "Process ID", optional = True)
])
@@ -34,17 +34,17 @@ class Volshell(generic.Volshell):
return
print(f"No task with task ID {pid} found")
def list_tasks(self):
def list_tasks(self, method = None):
"""Returns a list of task objects from the primary layer"""
# We always use the main kernel memory and associated symbols
return list(pslist.PsList.list_tasks(self.context, self.config['primary'], self.config['darwin']))
return list(pslist.PsList.get_list_tasks(method)(self.context, self.current_kernel_name))
def construct_locals(self) -> List[Tuple[List[str], Any]]:
result = super().construct_locals()
result += [
(['ct', 'change_task', 'cp'], self.change_task),
(['lt', 'list_tasks', 'ps'], self.list_tasks),
(['symbols'], self.context.symbol_space[self.config['darwin']]),
(['symbols'], self.context.symbol_space[self.current_symbol_table]),
]
if self.config.get('pid', None) is not None:
self.change_task(self.config['pid'])
@@ -62,5 +62,11 @@ class Volshell(generic.Volshell):
def display_symbols(self, symbol_table: str = None):
"""Prints an alphabetical list of symbols for a symbol table"""
if symbol_table is None:
symbol_table = self.config['darwin']
symbol_table = self.current_symbol_table
return super().display_symbols(symbol_table)
@property
def current_layer(self):
if self.__current_layer is None:
self.__current_layer = self.kernel.layer_name
return self.__current_layer
+13 -7
View File
@@ -5,7 +5,7 @@
from typing import Any, List, Tuple, Union
from volatility3.cli.volshell import generic
from volatility3.framework import interfaces, constants
from volatility3.framework import constants, interfaces
from volatility3.framework.configuration import requirements
from volatility3.plugins.windows import pslist
@@ -15,8 +15,8 @@ class Volshell(generic.Volshell):
@classmethod
def get_requirements(cls):
return (super().get_requirements() + [
requirements.SymbolTableRequirement(name = "nt_symbols", description = "Windows kernel symbols"),
return ([
requirements.ModuleRequirement(name = 'kernel', description = 'Windows kernel'),
requirements.PluginRequirement(name = 'pslist', plugin = pslist.PsList, version = (2, 0, 0)),
requirements.IntRequirement(name = 'pid', description = "Process ID", optional = True)
])
@@ -34,14 +34,14 @@ class Volshell(generic.Volshell):
def list_processes(self):
"""Returns a list of EPROCESS objects from the primary layer"""
# We always use the main kernel memory and associated symbols
return list(pslist.PsList.list_processes(self.context, self.config['primary'], self.config['nt_symbols']))
return list(pslist.PsList.list_processes(self.context, self.current_layer, self.current_symbol_table))
def construct_locals(self) -> List[Tuple[List[str], Any]]:
result = super().construct_locals()
result += [
(['cp', 'change_process'], self.change_process),
(['lp', 'list_processes', 'ps'], self.list_processes),
(['symbols'], self.context.symbol_space[self.config['nt_symbols']]),
(['symbols'], self.context.symbol_space[self.current_symbol_table]),
]
if self.config.get('pid', None) is not None:
self.change_process(self.config['pid'])
@@ -53,11 +53,17 @@ class Volshell(generic.Volshell):
"""Display Type describes the members of a particular object in alphabetical order"""
if isinstance(object, str):
if constants.BANG not in object:
object = self.config['nt_symbols'] + constants.BANG + object
object = self.current_symbol_table + constants.BANG + object
return super().display_type(object, offset)
def display_symbols(self, symbol_table: str = None):
"""Prints an alphabetical list of symbols for a symbol table"""
if symbol_table is None:
symbol_table = self.config['nt_symbols']
symbol_table = self.current_symbol_table
return super().display_symbols(symbol_table)
@property
def current_layer(self):
if self.__current_layer is None:
self.__current_layer = self.kernel.layer_name
return self.__current_layer
+6 -15
View File
@@ -21,14 +21,6 @@ from volatility3.framework.configuration import requirements
vollog = logging.getLogger(__name__)
windows_automagic = [
'ConstructionMagic', 'LayerStacker', 'KernelPDBScanner', 'WinSwapLayers', 'KernelModule'
]
linux_automagic = ['ConstructionMagic', 'LayerStacker', 'LinuxBannerCache', 'LinuxSymbolFinder', 'KernelModule']
mac_automagic = ['ConstructionMagic', 'LayerStacker', 'MacBannerCache', 'MacSymbolFinder', 'KernelModule']
def available(context: interfaces.context.ContextInterface) -> List[interfaces.automagic.AutomagicInterface]:
"""Returns an ordered list of all subclasses of
@@ -58,10 +50,7 @@ def choose_automagic(
plugin_category = "None"
plugin_categories = plugin.__module__.split('.')
lowest_index = len(plugin_categories)
automagic_categories = {'windows': windows_automagic, 'linux': linux_automagic, 'mac': mac_automagic}
for os in automagic_categories:
for os in constants.OS_CATEGORIES:
try:
if plugin_categories.index(os) < lowest_index:
lowest_index = plugin_categories.index(os)
@@ -70,14 +59,16 @@ def choose_automagic(
# The value wasn't found, try the next one
pass
if plugin_category not in automagic_categories:
if plugin_category not in constants.OS_CATEGORIES:
vollog.info("No plugin category detected")
return automagics
vollog.info(f"Detected a {plugin_category} category plugin")
output = []
for amagic in automagics:
if amagic.__class__.__name__ in automagic_categories[plugin_category]:
if plugin_category not in amagic.exclusion_list:
# Only include uncategorized automagic, or platform specific automagic
# (This allows user defined/uncategorized automagic to be included)
output += [amagic]
return output
+8
View File
@@ -45,6 +45,12 @@ class LinuxIntelStacker(interfaces.automagic.StackerLayerInterface):
symbol_files = linux_banners.get(banner, None)
if symbol_files:
if len(symbol_files) > 1:
using = "*"
vollog.warning(f"Multiple symbol files identified (using {using}):")
for symbol_file in symbol_files:
vollog.warning(f" {using} {symbol_file}")
using = " "
isf_path = symbol_files[0]
table_name = context.symbol_space.free_table_name('LintelStacker')
table = linux.LinuxKernelIntermedSymbols(context,
@@ -147,6 +153,7 @@ class LinuxBannerCache(symbol_cache.SymbolBannerCache):
os = "linux"
symbol_name = "linux_banner"
banner_path = constants.LINUX_BANNERS_PATH
exclusion_list = ['mac', 'windows']
class LinuxSymbolFinder(symbol_finder.SymbolFinder):
@@ -156,3 +163,4 @@ class LinuxSymbolFinder(symbol_finder.SymbolFinder):
banner_cache = LinuxBannerCache
symbol_class = "volatility3.framework.symbols.linux.LinuxKernelIntermedSymbols"
find_aslr = lambda cls, *args: LinuxIntelStacker.find_aslr(*args)[1]
exclusion_list = ['mac', 'windows']
+2
View File
@@ -202,6 +202,7 @@ class MacBannerCache(symbol_cache.SymbolBannerCache):
os = "mac"
symbol_name = "version"
banner_path = constants.MAC_BANNERS_PATH
exclusion_list = ['windows', 'linux']
class MacSymbolFinder(symbol_finder.SymbolFinder):
@@ -211,3 +212,4 @@ class MacSymbolFinder(symbol_finder.SymbolFinder):
banner_cache = MacBannerCache
find_aslr = MacIntelStacker.find_aslr
symbol_class = "volatility3.framework.symbols.mac.MacKernelIntermedSymbols"
exclusion_list = ['windows', 'linux']
+15 -2
View File
@@ -44,6 +44,7 @@ class KernelPDBScanner(interfaces.automagic.AutomagicInterface):
"""
priority = 30
max_pdb_size = 0x400000
exclusion_list = ['linux', 'mac']
def find_virtual_layers_from_req(self, context: interfaces.context.ContextInterface, config_path: str,
requirement: interfaces.configuration.RequirementInterface) -> List[str]:
@@ -145,8 +146,13 @@ class KernelPDBScanner(interfaces.automagic.AutomagicInterface):
return None
return (virtual_layer_name, kernel['mz_offset'], kernel)
vollog.debug("Kernel base determination - optimized scan virtual layer")
valid_kernel = self._method_layer_pdb_scan(context, vlayer, test_virtual_kernel, True, False, progress_callback)
if valid_kernel != None:
return valid_kernel
vollog.debug("Kernel base determination - slow scan virtual layer")
return self._method_layer_pdb_scan(context, vlayer, test_virtual_kernel, False, progress_callback)
return self._method_layer_pdb_scan(context, vlayer, test_virtual_kernel, False, False, progress_callback)
def method_fixed_mapping(self,
context: interfaces.context.ContextInterface,
@@ -174,12 +180,13 @@ class KernelPDBScanner(interfaces.automagic.AutomagicInterface):
vollog.debug(f"Potential kernel_virtual_offset caused a page fault: {hex(kvo)}")
vollog.debug("Kernel base determination - testing fixed base address")
return self._method_layer_pdb_scan(context, vlayer, test_physical_kernel, True, progress_callback)
return self._method_layer_pdb_scan(context, vlayer, test_physical_kernel, False, True, progress_callback)
def _method_layer_pdb_scan(self,
context: interfaces.context.ContextInterface,
vlayer: layers.intel.Intel,
test_kernel: Callable,
optimized: bool = False,
physical: bool = True,
progress_callback: constants.ProgressCallback = None) -> Optional[ValidKernelType]:
# TODO: Verify this is a windows image
@@ -191,9 +198,15 @@ class KernelPDBScanner(interfaces.automagic.AutomagicInterface):
if not physical:
layer_to_scan = virtual_layer_name
start_scan_address = 0
if optimized and not physical and context.layers[layer_to_scan].metadata.architecture in ["Intel64"]:
# TODO: change this value accordingly when 5-Level paging is supported.
start_scan_address = (0x1f0 << 39)
kernel_pdb_names = [bytes(name + ".pdb", "utf-8") for name in constants.windows.KERNEL_MODULE_NAMES]
kernels = PDBUtility.pdbname_scan(ctx = context,
layer_name = layer_to_scan,
start = start_scan_address,
page_size = vlayer.page_size,
pdb_names = kernel_pdb_names,
progress_callback = progress_callback)
+61 -22
View File
@@ -28,9 +28,9 @@ The self-referential indices for older versions of windows are listed below:
"""
import logging
import struct
from typing import Generator, List, Optional, Tuple, Type, Iterable
from typing import Generator, Iterable, List, Optional, Tuple, Type
from volatility3.framework import interfaces, layers, constants
from volatility3.framework import constants, interfaces, layers
from volatility3.framework.configuration import requirements
from volatility3.framework.layers import intel
@@ -116,10 +116,27 @@ class DtbSelfRefPae(DtbSelfReferential):
mask = 0x3FFFFFFFFFF000,
reserved_bits = 0x0)
def __call__(self, *args, **kwargs):
dtb = super().__call__(*args, **kwargs)
@staticmethod
def _and_bytes(abytes, bbytes):
return bytes([a & b for a, b in zip(abytes[::-1], bbytes[::-1])][::-1])
def __call__(self, data: bytes, data_offset: int, page_offset: int) -> Optional[Tuple[int, int]]:
dtb = super().__call__(data, data_offset, page_offset)
if dtb:
return dtb[0] - 0x4000, dtb[1]
# Find the top page
top_pae_page = dtb[0] - 0x4000
# The top page should map to the next four pages after it
# Build what we expect the page table to be
expected_table = b''.join([struct.pack(self.ptr_struct, top_pae_page + (i * 0x1000)) for i in range(1, 5)])
# Mask off the page bits of top level page map
page_table_mask = b"\x00\xf0\xff\xff\xff\xff\xff\xff" * 4
page_table = data[top_pae_page - data_offset: top_pae_page - data_offset + (4 * self.ptr_size)]
# Compare them
anded_bytes = self._and_bytes(page_table, page_table_mask)
if (anded_bytes == expected_table):
return top_pae_page, dtb[1]
# Return None since the dtb value *isn't* None
return None
return dtb
@@ -202,30 +219,50 @@ class WindowsIntelStacker(interfaces.automagic.StackerLayerInterface):
for description, tests, sections in cls.test_sets:
vollog.debug(description)
# There is a very high chance that the DTB will live in these very narrow segments, assuming we couldn't find them previously
hits = context.layers[layer_name].scan(context,
PageMapScanner(tests = tests),
sections = sections,
progress_callback = progress_callback)
hits = base_layer.scan(context,
PageMapScanner(tests = tests),
sections = sections,
progress_callback = progress_callback)
# Flatten the generator
def sort_by_tests(x):
"""Key used to sort by tests"""
return tests.index(x[0]), x[1]
def get_max_pointer(page_table, test, ptr_size: int):
"""Determines a pointer from a page_table"""
max_ptr = 0
for index in range(0, len(page_table), ptr_size):
pointer = struct.unpack(test.ptr_struct, page_table[index:index + ptr_size])[0]
# Make sure the pointer is valid, ignore large pages which would require more calculation
if pointer & 0x1 and not pointer & 0x80:
max_ptr = max(max_ptr, (pointer ^ (pointer & 0xfff)) % test.layer_type.maximum_address)
return max_ptr
hits = sorted(list(hits), key = sort_by_tests)
if hits:
# TODO: Decide which to use if there are multiple options
test, page_map_offset = hits[0]
vollog.debug(f"{test.__class__.__name__} test succeeded at {hex(page_map_offset)}")
new_layer_name = context.layers.free_layer_name("IntelLayer")
config_path = interfaces.configuration.path_join("IntelHelper", new_layer_name)
context.config[interfaces.configuration.path_join(config_path, "memory_layer")] = layer_name
context.config[interfaces.configuration.path_join(config_path, "page_map_offset")] = page_map_offset
# TODO: Need to determine the layer type (chances are high it's x64, hence this default)
layer = test.layer_type(context,
config_path = config_path,
name = new_layer_name,
metadata = {'os': 'Windows'})
for test, page_map_offset in hits:
# Turn the page tables into integers and find the largest one
page_table = base_layer.read(page_map_offset, 0x1000)
ptr_size = struct.calcsize(test.ptr_struct)
max_pointer = get_max_pointer(page_table, test, ptr_size)
if max_pointer <= base_layer.maximum_address:
vollog.debug(f"{test.__class__.__name__} test succeeded at {hex(page_map_offset)}")
new_layer_name = context.layers.free_layer_name("IntelLayer")
config_path = interfaces.configuration.path_join("IntelHelper", new_layer_name)
context.config[interfaces.configuration.path_join(config_path, "memory_layer")] = layer_name
context.config[
interfaces.configuration.path_join(config_path, "page_map_offset")] = page_map_offset
layer = test.layer_type(context,
config_path = config_path,
name = new_layer_name,
metadata = {'os': 'Windows'})
break
else:
vollog.debug(
f"Max pointer for hit with test {test.__class__.__name__} not met: {hex(max_pointer)} > {hex(base_layer.maximum_address)}")
if layer is not None and config_path:
break
if layer is not None and config_path:
@@ -238,6 +275,8 @@ class WinSwapLayers(interfaces.automagic.AutomagicInterface):
"""Class to read swap_layers filenames from single-swap-layers, create the
layers and populate the single-layers swap_layers."""
exclusion_list = ['linux', 'mac']
def __call__(self,
context: interfaces.context.ContextInterface,
config_path: str,
@@ -10,7 +10,7 @@ expect to be in the context (such as particular layers or symboltables).
"""
import abc
import logging
from typing import Any, ClassVar, List, Optional, Type, Dict, Tuple
from typing import Any, ClassVar, Dict, List, Optional, Tuple, Type
from volatility3.framework import constants, interfaces
@@ -303,7 +303,8 @@ class TranslationLayerRequirement(interfaces.configuration.ConstructableRequirem
args = {"context": context, "config_path": config_path, "name": name}
if any(
[subreq.unsatisfied(context, config_path) for subreq in self.requirements.values() if not subreq.optional]):
[subreq.unsatisfied(context, config_path) for subreq in self.requirements.values() if
not subreq.optional]):
return None
obj = self._construct_class(context, config_path, args)
@@ -358,7 +359,8 @@ class SymbolTableRequirement(interfaces.configuration.ConstructableRequirementIn
args = {"context": context, "config_path": config_path, "name": name}
if any(
[subreq.unsatisfied(context, config_path) for subreq in self.requirements.values() if not subreq.optional]):
[subreq.unsatisfied(context, config_path) for subreq in self.requirements.values() if
not subreq.optional]):
return None
# Fill out the parameter for class creation
@@ -462,6 +464,15 @@ class ModuleRequirement(interfaces.configuration.ConstructableRequirementInterfa
"TypeError - Module Requirement only accepts string labels: {}".format(repr(value)))
return {config_path: self}
result = {}
for subreq in self._requirements:
req_unsatisfied = self._requirements[subreq].unsatisfied(context, config_path)
if req_unsatisfied:
result.update(req_unsatisfied)
if not result:
result = {config_path: self}
return result
### NOTE: This validate method has side effects (the dependencies can change)!!!
self._validate_class(context, interfaces.configuration.parent_path(config_path))
@@ -482,7 +493,8 @@ class ModuleRequirement(interfaces.configuration.ConstructableRequirementInterfa
args = {"context": context, "config_path": config_path, "name": name}
if any(
[subreq.unsatisfied(context, config_path) for subreq in self.requirements.values() if not subreq.optional]):
[subreq.unsatisfied(context, config_path) for subreq in self.requirements.values() if
not subreq.optional]):
return None
obj = self._construct_class(context, config_path, args)
+5 -3
View File
@@ -9,7 +9,7 @@ volatility This includes default scanning block sizes, etc.
import enum
import os.path
import sys
from typing import Optional, Callable
from typing import Callable, Optional
import volatility3.framework.constants.linux
import volatility3.framework.constants.windows
@@ -39,7 +39,7 @@ BANG = "!"
# We use the SemVer 2.0.0 versioning scheme
VERSION_MAJOR = 2 # Number of releases of the library with a breaking change
VERSION_MINOR = 0 # Number of changes that only add to the interface
VERSION_MINOR = 1 # Number of changes that only add to the interface
VERSION_PATCH = 0 # Number of changes that do not change the interface
VERSION_SUFFIX = ""
@@ -63,7 +63,7 @@ LOGLEVEL_VVVV = 6
CACHE_PATH = os.path.join(os.path.expanduser("~"), ".cache", "volatility3")
"""Default path to store cached data"""
if sys.platform == 'windows':
if sys.platform == 'win32':
CACHE_PATH = os.path.join(os.environ.get("APPDATA", os.path.expanduser("~")), "volatility3")
os.makedirs(CACHE_PATH, exist_ok = True)
@@ -78,6 +78,8 @@ BUG_URL = "https://github.com/volatilityfoundation/volatility3/issues"
ProgressCallback = Optional[Callable[[float, str], None]]
"""Type information for ProgressCallback objects"""
OS_CATEGORIES = ['windows', 'mac', 'linux']
class Parallelism(enum.IntEnum):
"""An enumeration listing the different types of parallelism applied to
@@ -1,10 +1,12 @@
# 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
#
"""Volatility 3 Linux Constants.
"""Volatility 3 Windows Constants.
Windows-specific values that aren't found in debug symbols
"""
KERNEL_MODULE_NAMES = ["ntkrnlmp", "ntkrnlpa", "ntkrpamp", "ntoskrnl"]
"""The list of names that kernel modules can have within the windows OS"""
PE_MAX_EXTRACTION_SIZE = 1024 * 1024 * 256
+1 -1
View File
@@ -141,7 +141,7 @@ class Context(interfaces.context.ContextInterface):
layer_name: The layer within the context in which the module exists
offset: The offset at which the module exists in the layer
native_layer_name: The default native layer for objects constructed by the module
size: The size, in bytes, that the module occupys from offset location within the layer named layer_name
size: The size, in bytes, that the module occupies from offset location within the layer named layer_name
"""
if size:
return SizedModule.create(self,
@@ -40,6 +40,9 @@ class AutomagicInterface(interfaces.configuration.ConfigurableInterface, metacla
priority = 10
"""An ordering to indicate how soon this automagic should be run"""
exclusion_list = []
"""A list of plugin categories (typically operating systems) which the plugin will not operate on"""
def __init__(self, context: interfaces.context.ContextInterface, config_path: str, *args, **kwargs) -> None:
super().__init__(context, config_path)
for requirement in self.get_requirements():
@@ -73,7 +73,7 @@ class HierarchicalDict(collections.abc.Mapping):
separator: str = CONFIG_SEPARATOR) -> None:
"""
Args:
initial_dict: A dictionary to populate the HierachicalDict with initially
initial_dict: A dictionary to populate the HierarchicalDict with initially
separator: A custom hierarchy separator (defaults to CONFIG_SEPARATOR)
"""
if not (isinstance(separator, str) and len(separator) == 1):
+1 -1
View File
@@ -129,7 +129,7 @@ class ContextInterface(metaclass = ABCMeta):
layer_name: The layer the module is associated with (which layer the module lives within)
offset: The initial/base offset of the module (used as the offset for relative symbols)
native_layer_name: The default native_layer_name to use when the module constructs objects
size: The size, in bytes, that the module occupys from offset location within the layer named layer_name
size: The size, in bytes, that the module occupies from offset location within the layer named layer_name
Returns:
A module object
+1 -1
View File
@@ -115,7 +115,7 @@ class ObjectInterface(metaclass = abc.ABCMeta):
mask = context.layers[object_info.layer_name].address_mask
normalized_offset = object_info.offset & mask
self._vol = collections.ChainMap({}, object_info, {'type_name': type_name, 'offset': normalized_offset}, kwargs)
self._vol = collections.ChainMap({}, {'type_name': type_name, 'offset': normalized_offset}, object_info, kwargs)
self._context = context
def __getattr__(self, attr: str) -> Any:
@@ -1,7 +1,7 @@
# 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
#
"""All plugins output a TreeGrid object which must then be rendered (eithe by a
"""All plugins output a TreeGrid object which must then be rendered (either by a
GUI, or as text output, html output or in some other form.
This module defines both the output format (:class:`TreeGrid`) and the
+24 -7
View File
@@ -54,7 +54,7 @@ class QemuSuspendLayer(segmented.NonLinearlySegmentedLayer):
def _read_configuration(self, base_layer: interfaces.layers.DataLayerInterface, name: str) -> Any:
"""Reads the JSON configuration from the end of the file"""
chunk_size = 0x4096
chunk_size = 4096
data = b''
for i in range(base_layer.maximum_address, base_layer.minimum_address, -chunk_size):
if i != base_layer.maximum_address:
@@ -65,6 +65,7 @@ class QemuSuspendLayer(segmented.NonLinearlySegmentedLayer):
if start_of_json >= 0:
data = data[start_of_json:]
return json.loads(data)
# No JSON configuration found at the end of the file, return empty dict
return dict()
raise exceptions.LayerException(name, "Invalid JSON configuration at the end of the file")
@@ -79,9 +80,11 @@ class QemuSuspendLayer(segmented.NonLinearlySegmentedLayer):
addr = self.context.object(self._qemu_table_name + constants.BANG + 'unsigned long long',
offset = index,
layer_name = self._base_layer)
# Flags are stored in the n least significant bits, where n equals the bit-length of pagesize
flags = addr & (page_size - 1)
page_size_bits = int(math.log(page_size, 2))
addr = (addr >> page_size_bits) << page_size_bits
# addr equals the highest multiple of pagesize <= offset
# (We assume that page_size is a power of 2)
addr = addr ^ (addr & (page_size - 1))
index += 8
if flags & self.SEGMENT_FLAG_MEM_SIZE:
@@ -126,6 +129,7 @@ class QemuSuspendLayer(segmented.NonLinearlySegmentedLayer):
self._configuration = self._read_configuration(base_layer, self.name)
section_byte = -1
index = 8
section_info = dict()
current_section_id = -1
version_id = -1
name = None
@@ -162,6 +166,8 @@ class QemuSuspendLayer(segmented.NonLinearlySegmentedLayer):
offset = index,
layer_name = self._base_layer)
index += 4
# Store section info for handling QEVM_SECTION_PARTs later on
section_info[current_section_id] = {'name': name, 'version_id': version_id}
# Read additional data
index = self.extract_data(index, name, version_id)
elif section_byte == self.QEVM_SECTION_PART or section_byte == self.QEVM_SECTION_END:
@@ -171,7 +177,8 @@ class QemuSuspendLayer(segmented.NonLinearlySegmentedLayer):
current_section_id = section_id
index += 4
# Read additional data
index = self.extract_data(index, name, version_id)
index = self.extract_data(index, section_info[current_section_id]['name'],
section_info[current_section_id]['version_id'])
elif section_byte == self.QEVM_SECTION_FOOTER:
section_id = self.context.object(self._qemu_table_name + constants.BANG + 'unsigned long',
offset = index,
@@ -189,7 +196,7 @@ class QemuSuspendLayer(segmented.NonLinearlySegmentedLayer):
if name == 'ram':
if version_id != 4:
raise exceptions.LayerException(f"QEMU unknown RAM version_id {version_id}")
new_segments, index = self._get_ram_segments(index, self._configuration.get('page_size', None) or 4096)
new_segments, index = self._get_ram_segments(index, self._configuration.get('page_size', 4096))
self._segments += new_segments
elif name == 'spapr/htab':
if version_id != 1:
@@ -208,6 +215,13 @@ class QemuSuspendLayer(segmented.NonLinearlySegmentedLayer):
layer_name = self._base_layer)
htab_index, htab_n_valid, htab_n_invalid = htab
index += 8 + (htab_n_valid * self.HASH_PTE_SIZE_64)
elif name == 'dirty-bitmap':
index += 1
elif name == 'pbs-state':
section_len = self.context.object(self._qemu_table_name + constants.BANG + 'unsigned long long',
offset = index,
layer_name = self._base_layer)
index += 8 + section_len
return index
def _decode_data(self, data: bytes, mapped_offset: int, offset: int, output_length: int) -> bytes:
@@ -217,9 +231,12 @@ class QemuSuspendLayer(segmented.NonLinearlySegmentedLayer):
of the starting data. It is the responsibility of the layer to turn the provided data chunk into the right
portion of data necessary.
"""
start_offset = offset ^ (offset & 0xfff)
page_size = self._configuration.get('page_size', 4096)
# start_offset equals the highest multiple of pagesize <= offset
# (We assume that page_size is a power of 2)
start_offset = offset ^ (offset & (page_size - 1))
if start_offset in self._compressed:
data = (data * 0x1000)
data = (data * page_size)
result = data[offset - start_offset:output_length + offset - start_offset]
return result
+17 -1
View File
@@ -10,10 +10,11 @@ import logging
import lzma
import os
import ssl
import sys
import urllib.parse
import urllib.request
import zipfile
from typing import Optional, Any, IO, List
from typing import Any, IO, List, Optional
from urllib import error
from volatility3 import framework
@@ -100,6 +101,21 @@ class ResourceAccessor(object):
"""
urllib.request.install_opener(urllib.request.build_opener(*self._handlers))
# Python bug 46654
if sys.platform == 'win32':
# We only need to worry about UNC paths on windows, on linux they'd be smb:// and need pysmb or similar
parsed_url = urllib.parse.urlparse(url, scheme = 'file')
# Only worry about file scheme URLs, make sure that there's either a host or
# the unparsing left an extra slash at the start (which will get lost with urlunparse)
if parsed_url.scheme == 'file' and (parsed_url.netloc or parsed_url.path.startswith('//')):
# Change the netloc to '/' and then prepend the netloc to the path
# Urlunparse will remove extra initial slashes from path, hence setting netloc
new_url = urllib.parse.urlunparse((parsed_url.scheme, '/',
'/' + parsed_url.netloc + parsed_url.path, parsed_url.params,
parsed_url.query, parsed_url.fragment))
vollog.log(constants.LOGLEVEL_VVVV, f'UNC path detected, converted path {url} to {new_url}')
url = new_url
try:
fp = urllib.request.urlopen(url, context = self._context)
except error.URLError as excp:
@@ -31,7 +31,7 @@ class BytesScanner(layers.ScannerInterface):
class RegExScanner(layers.ScannerInterface):
"""A scanner that can be provided with a bytes-object regular expression pattern
The scanner will scqn all blocks for the regular expression and report the absolute offset of any finds
The scanner will scan all blocks for the regular expression and report the absolute offset of any finds
The default flags include DOTALL, since the searches are through binary data and the newline character should
have no specific significance in such searches"""
@@ -95,7 +95,7 @@ class MultiStringScanner(layers.ScannerInterface):
else:
suffixes.append(re.escape(bytes([entry])))
else:
# If we've fininshed one of the strings at this point, remember it for later
# If we've finished one of the strings at this point, remember it for later
finished = True
if len(suffixes) == 1:
+9 -6
View File
@@ -6,9 +6,9 @@ import collections
import collections.abc
import logging
import struct
from typing import Any, ClassVar, Dict, List, Iterable, Optional, Tuple, Type, Union as TUnion, overload
from typing import Any, ClassVar, Dict, Iterable, List, Optional, Tuple, Type, Union as TUnion, overload
from volatility3.framework import interfaces, constants
from volatility3.framework import constants, interfaces
from volatility3.framework.objects import templates, utility
vollog = logging.getLogger(__name__)
@@ -136,12 +136,15 @@ class PrimitiveObject(interfaces.objects.ObjectInterface):
if k not in ["context", "data_format", "object_info", "type_name"]:
kwargs[k] = v
kwargs['new_value'] = self.__new_value
return (self._context, self._vol.maps[-2]['type_name'], self._vol.maps[-3], self._data_format), kwargs
return (self._context, self._vol.maps[-3]['type_name'], self._vol.maps[-2], self._data_format), kwargs
@classmethod
def _unmarshall(cls, context: interfaces.context.ContextInterface, data_format: DataFormatInfo,
object_info: interfaces.objects.ObjectInformation) -> TUnion[int, float, bool, bytes, str]:
data = context.layers.read(object_info.layer_name, object_info.offset, data_format.length)
# Don't try to lookup a 0 length data format, incase it's at an invalid offset. Length 0 means b''
data = b''
if data_format.length > 0:
data = context.layers.read(object_info.layer_name, object_info.offset, data_format.length)
return convert_data_to_value(data, cls._struct_type, data_format)
class VolTemplateProxy(interfaces.objects.ObjectInterface.VolTemplateProxy):
@@ -203,7 +206,7 @@ class Bytes(PrimitiveObject, bytes):
length: int = 1,
**kwargs) -> 'Bytes':
"""Creates the appropriate class and returns it so that the native type
is inherritted.
is inherited.
The only reason the kwargs is added, is so that the
inheriting types can override __init__ without needing to
@@ -701,7 +704,7 @@ class AggregateType(interfaces.objects.ObjectInterface):
tmp_list[member] = (relative_offset, new_child)
# If there's trouble with mutability, consider making update_vol return a clone with the changes
# (there will be a few other places that will be necessary) and/or making these part of the
# permanent dictionaries rather than the non-clonable ones
# permanent dictionaries rather than the non-cloneable ones
template.update_vol(members = tmp_list)
@classmethod
+2 -2
View File
@@ -110,9 +110,9 @@ class LayerWriter(plugins.PluginInterface):
def _generate_layers(self):
"""List layer names from this run"""
for name in self.context.layers:
yield (0, (name, ))
yield (0, (name, self.context.layers[name].__class__.__name__))
def run(self):
if self.config['list']:
return renderers.TreeGrid([("Layer name", str)], self._generate_layers())
return renderers.TreeGrid([("Layer name", str), ('Layer type', str)], self._generate_layers())
return renderers.TreeGrid([("Status", str)], self._generator())
+38 -28
View File
@@ -4,9 +4,9 @@
import logging
from abc import ABC, abstractmethod
from enum import Enum
from typing import List, Iterator, Tuple, Generator
from typing import Generator, Iterator, List, Tuple
from volatility3.framework import renderers, interfaces, constants, contexts, class_subclasses
from volatility3.framework import class_subclasses, constants, contexts, interfaces, renderers
from volatility3.framework.configuration import requirements
from volatility3.framework.interfaces import plugins
from volatility3.framework.objects import utility
@@ -15,39 +15,39 @@ vollog = logging.getLogger(__name__)
class DescStateEnum(Enum):
desc_miss = -1 # ID mismatch (pseudo state)
desc_reserved = 0x0 # reserved, in use by writer
desc_committed = 0x1 # committed by writer, could get reopened
desc_finalized = 0x2 # committed, no further modification allowed
desc_reusable = 0x3 # free, not yet used by any writer
desc_miss = -1 # ID mismatch (pseudo state)
desc_reserved = 0x0 # reserved, in use by writer
desc_committed = 0x1 # committed by writer, could get reopened
desc_finalized = 0x2 # committed, no further modification allowed
desc_reusable = 0x3 # free, not yet used by any writer
class ABCKmsg(ABC):
"""Kernel log buffer reader"""
LEVELS = (
"emerg", # system is unusable
"alert", # action must be taken immediately
"crit", # critical conditions
"err", # error conditions
"warn", # warning conditions
"notice", # normal but significant condition
"info", # informational
"debug", # debug-level messages
"emerg", # system is unusable
"alert", # action must be taken immediately
"crit", # critical conditions
"err", # error conditions
"warn", # warning conditions
"notice", # normal but significant condition
"info", # informational
"debug", # debug-level messages
)
FACILITIES = (
"kern", # kernel messages
"user", # random user-level messages
"mail", # mail system
"daemon", # system daemons
"auth", # security/authorization messages
"syslog", # messages generated internally by syslogd
"lpr", # line printer subsystem
"news", # network news subsystem
"uucp", # UUCP subsystem
"cron", # clock daemon
"kern", # kernel messages
"user", # random user-level messages
"mail", # mail system
"daemon", # system daemons
"auth", # security/authorization messages
"syslog", # messages generated internally by syslogd
"lpr", # line printer subsystem
"news", # network news subsystem
"uucp", # UUCP subsystem
"cron", # clock daemon
"authpriv", # security/authorization messages (private)
"ftp" # FTP daemon
"ftp" # FTP daemon
)
def __init__(
@@ -247,12 +247,20 @@ class KmsgFiveTen(ABCKmsg):
The data block ring 'text_data_ring' contains the records' text strings.
A pointer to the high level structure is kept in the prb pointer which is
initialized to a static ringbuffer.
.. code-block:: c
static struct printk_ringbuffer *prb = &printk_rb_static;
In SMP systems with more than 64 CPUs this ringbuffer size is dynamically
allocated according the number of CPUs based on the value of
CONFIG_LOG_CPU_MAX_BUF_SHIFT. The prb pointer is updated consequently to
this dynamic ringbuffer in setup_log_buf().
.. code-block:: c
prb = &printk_rb_dynamic;
Behind scenes, log_buf is still used as external buffer.
When the static printk_ringbuffer struct is initialized, _DEFINE_PRINTKRB
sets text_data_ring.data pointer to the address in log_buf which points to
@@ -262,12 +270,14 @@ class KmsgFiveTen(ABCKmsg):
buffer via the prb_init function.
In that case, the original external static buffer in __log_buf and
printk_rb_static are unused.
...
.. code-block:: c
new_log_buf = memblock_alloc(new_log_buf_len, LOG_ALIGN);
prb_init(&printk_rb_dynamic, new_log_buf, ...);
log_buf = new_log_buf;
prb = &printk_rb_dynamic;
...
See printk.c and printk_ringbuffer.c in kernel/printk/ folder for more
details.
"""
@@ -0,0 +1,8 @@
# This file is Copyright 2022 Volatility Foundation and licensed under the Volatility Software License 1.0
# which is available at https://www.volatilityfoundation.org/license/vsl-v1.0
#
"""All core mac plugins.
These modules should only be imported from volatility3.plugins NOT
volatility3.framework.plugins
"""
@@ -9,7 +9,7 @@ from volatility3.framework.symbols import mac
class Ifconfig(plugins.PluginInterface):
"""Lists loaded kernel modules"""
"""Lists network interface information for all devices"""
_required_framework_version = (2, 0, 0)
@@ -1,4 +1,4 @@
# This file is opyright 2020 Volatility Foundation and licensed under the Volatility Software License 1.0
# This file is Copyright 2020 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
+1 -1
View File
@@ -1,4 +1,4 @@
# This file is opyright 2020 Volatility Foundation and licensed under the Volatility Software License 1.0
# This file is Copyright 2020 Volatility Foundation and licensed under the Volatility Software License 1.0
# which is available at https://www.volatilityfoundation.org/license/vsl-v1.0
#
+1 -1
View File
@@ -12,7 +12,7 @@ from volatility3.framework.symbols import mac
class Mount(plugins.PluginInterface):
"""A module containing a collection of plugins that produce data typically
foundin Mac's mount command"""
found in Mac's mount command"""
_required_framework_version = (2, 0, 0)
@@ -1,4 +1,4 @@
# This file is opyright 2020 Volatility Foundation and licensed under the Volatility Software License 1.0
# This file is Copyright 2020 Volatility Foundation and licensed under the Volatility Software License 1.0
# which is available at https://www.volatilityfoundation.org/license/vsl-v1.0
#
+27 -18
View File
@@ -12,7 +12,7 @@ import traceback
from typing import Generator, Iterable, List, Optional, Tuple, Type
from volatility3 import framework
from volatility3.framework import renderers, automagic, interfaces, plugins, exceptions
from volatility3.framework import automagic, exceptions, interfaces, plugins, renderers
from volatility3.framework.configuration import requirements
vollog = logging.getLogger(__name__)
@@ -74,10 +74,6 @@ class Timeliner(interfaces.plugins.PluginInterface):
@classmethod
def get_requirements(cls) -> List[interfaces.configuration.RequirementInterface]:
return [
requirements.StringRequirement(name = 'plugins',
description = "Comma separated list of plugins to run",
optional = True,
default = None),
requirements.BooleanRequirement(
name = 'record-config',
description = "Whether to record the state of all the plugins once complete",
@@ -110,6 +106,15 @@ class Timeliner(interfaces.plugins.PluginInterface):
row from each plugin."""
# Generate the results for each plugin
data = []
# Open the bodyfile now, so we can start outputting to it immediately
if self.config.get('create-bodyfile', True):
file_data = self.open("volatility.body")
fp = io.TextIOWrapper(file_data, write_through = True)
else:
file_data = None
fp = None
for plugin in runable_plugins:
plugin_name = plugin.__class__.__name__
self._progress_callback((runable_plugins.index(plugin) * 100) // len(runable_plugins),
@@ -130,27 +135,31 @@ class Timeliner(interfaces.plugins.PluginInterface):
times.get(TimeLinerType.ACCESSED, renderers.NotApplicableValue()),
times.get(TimeLinerType.CHANGED, renderers.NotApplicableValue())
]))
except Exception:
vollog.log(logging.INFO, f"Exception occurred running plugin: {plugin_name}")
vollog.log(logging.DEBUG, traceback.format_exc())
for data_item in sorted(data, key = self._sort_function):
yield data_item
# Write out a body file if necessary
if self.config.get('create-bodyfile', True):
with self.open("volatility.body") as file_data:
with io.TextIOWrapper(file_data, write_through = True) as fp:
for (plugin_name, item) in self.timeline:
# Write each entry because the body file doesn't need to be sorted
if fp:
times = self.timeline[(plugin_name, item)]
# Body format is: MD5|name|inode|mode_as_string|UID|GID|size|atime|mtime|ctime|crtime
if self._any_time_present(times):
fp.write("|{} - {}||||||{}|{}|{}|{}\n".format(
fp.write("|{} - {}|0|0|0|0|0|{}|{}|{}|{}\n".format(
plugin_name, self._sanitize_body_format(item),
self._text_format(times.get(TimeLinerType.ACCESSED, "")),
self._text_format(times.get(TimeLinerType.MODIFIED, "")),
self._text_format(times.get(TimeLinerType.CHANGED, "")),
self._text_format(times.get(TimeLinerType.CREATED, ""))))
except Exception:
vollog.log(logging.INFO, f"Exception occurred running plugin: {plugin_name}")
vollog.log(logging.DEBUG, traceback.format_exc())
for data_item in sorted(data, key = self._sort_function):
yield data_item
# Write out a body file if necessary
if self.config.get('create-bodyfile', True):
if fp:
fp.close()
file_data.close()
def _sanitize_body_format(self, value):
return value.replace("|", "_")
@@ -164,7 +173,7 @@ class Timeliner(interfaces.plugins.PluginInterface):
def _text_format(self, value):
"""Formats a value as text, in case it is an AbsentValue"""
if isinstance(value, interfaces.renderers.BaseAbsentValue):
return ""
return "0"
if isinstance(value, datetime.datetime):
return int(value.timestamp())
return value
@@ -202,7 +211,7 @@ class Timeliner(interfaces.plugins.PluginInterface):
if isinstance(plugin, TimeLinerInterface):
if not len(filter_list) or any(
[filter in plugin.__module__ + '.' + plugin.__class__.__name__ for filter in filter_list]):
[filter in plugin.__module__ + '.' + plugin.__class__.__name__ for filter in filter_list]):
plugins_to_run.append(plugin)
except exceptions.UnsatisfiedException as excp:
# Remove the failed plugin from the list and continue
@@ -21,7 +21,7 @@ class BigPools(interfaces.plugins.PluginInterface):
"""List big page pools."""
_required_framework_version = (2, 0, 0)
_version = (1, 0, 0)
_version = (1, 1, 0)
@classmethod
def get_requirements(cls) -> List[interfaces.configuration.RequirementInterface]:
@@ -32,7 +32,11 @@ class BigPools(interfaces.plugins.PluginInterface):
requirements.StringRequirement(name = 'tags',
description = "Comma separated list of pool tags to filter pools returned",
optional = True,
default = None)
default = None),
requirements.BooleanRequirement(name = 'show-free',
description = 'Show freed regions (otherwise only show allocations in use)',
default = False,
optional = True)
]
@classmethod
@@ -40,7 +44,8 @@ class BigPools(interfaces.plugins.PluginInterface):
context: interfaces.context.ContextInterface,
layer_name: str,
symbol_table: str,
tags: Optional[list] = None):
tags: Optional[list] = None,
show_free: bool = False):
"""Returns the big page pool objects from the kernel PoolBigPageTable array.
Args:
@@ -97,7 +102,7 @@ class BigPools(interfaces.plugins.PluginInterface):
for big_pool in big_pools:
if big_pool.is_valid():
if tags is None or big_pool.get_key() in tags:
if (tags is None or big_pool.get_key() in tags) and (show_free or not big_pool.is_free()):
yield big_pool
def _generator(self) -> Iterator[Tuple[int, Tuple[int, str]]]: # , str, int]]]:
@@ -110,13 +115,19 @@ class BigPools(interfaces.plugins.PluginInterface):
for big_pool in self.list_big_pools(context = self.context,
layer_name = kernel.layer_name,
symbol_table = kernel.symbol_table_name,
tags = tags):
tags = tags,
show_free = self.config.get("show-free")):
num_bytes = big_pool.get_number_of_bytes()
if not isinstance(num_bytes, interfaces.renderers.BaseAbsentValue):
num_bytes = format_hints.Hex(num_bytes)
yield (0, (format_hints.Hex(big_pool.Va), big_pool.get_key(), big_pool.get_pool_type(), num_bytes))
if big_pool.is_free():
status = "Free"
else:
status = "Allocated"
yield (0, (format_hints.Hex(big_pool.Va), big_pool.get_key(), big_pool.get_pool_type(), num_bytes, status))
def run(self):
return renderers.TreeGrid([
@@ -124,4 +135,5 @@ class BigPools(interfaces.plugins.PluginInterface):
('Tag', str),
('PoolType', str),
('NumberOfBytes', format_hints.Hex),
('Status', str),
], self._generator())
@@ -0,0 +1,167 @@
# This file is Copyright 2022 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 Iterator, List, Tuple
from volatility3.framework import constants, renderers, exceptions, interfaces
from volatility3.framework.configuration import requirements
from volatility3.framework.renderers import format_hints
from volatility3.plugins.windows import driverscan
DEVICE_CODES = {
0x00000027 : "FILE_DEVICE_8042_PORT",
0x00000032 : "FILE_DEVICE_ACPI",
0x00000029 : "FILE_DEVICE_BATTERY",
0x00000001 : "FILE_DEVICE_BEEP",
0x0000002a : "FILE_DEVICE_BUS_EXTENDER",
0x00000002 : "FILE_DEVICE_CD_ROM",
0x00000003 : "FILE_DEVICE_CD_ROM_FILE_SYSTEM",
0x00000030 : "FILE_DEVICE_CHANGER",
0x00000004 : "FILE_DEVICE_CONTROLLER",
0x00000005 : "FILE_DEVICE_DATALINK",
0x00000006 : "FILE_DEVICE_DFS",
0x00000035 : "FILE_DEVICE_DFS_FILE_SYSTEM",
0x00000036 : "FILE_DEVICE_DFS_VOLUME",
0x00000007 : "FILE_DEVICE_DISK",
0x00000008 : "FILE_DEVICE_DISK_FILE_SYSTEM",
0x00000033 : "FILE_DEVICE_DVD",
0x00000009 : "FILE_DEVICE_FILE_SYSTEM",
0x0000003a : "FILE_DEVICE_FIPS",
0x00000034 : "FILE_DEVICE_FULLSCREEN_VIDEO",
0x0000000a : "FILE_DEVICE_INPORT_PORT",
0x0000000b : "FILE_DEVICE_KEYBOARD",
0x0000002f : "FILE_DEVICE_KS",
0x00000039 : "FILE_DEVICE_KSEC",
0x0000000c : "FILE_DEVICE_MAILSLOT",
0x0000002d : "FILE_DEVICE_MASS_STORAGE",
0x0000000d : "FILE_DEVICE_MIDI_IN",
0x0000000e : "FILE_DEVICE_MIDI_OUT",
0x0000002b : "FILE_DEVICE_MODEM",
0x0000000f : "FILE_DEVICE_MOUSE",
0x00000010 : "FILE_DEVICE_MULTI_UNC_PROVIDER",
0x00000011 : "FILE_DEVICE_NAMED_PIPE",
0x00000012 : "FILE_DEVICE_NETWORK",
0x00000013 : "FILE_DEVICE_NETWORK_BROWSER",
0x00000014 : "FILE_DEVICE_NETWORK_FILE_SYSTEM",
0x00000028 : "FILE_DEVICE_NETWORK_REDIRECTOR",
0x00000015 : "FILE_DEVICE_NULL",
0x00000016 : "FILE_DEVICE_PARALLEL_PORT",
0x00000017 : "FILE_DEVICE_PHYSICAL_NETCARD",
0x00000018 : "FILE_DEVICE_PRINTER",
0x00000019 : "FILE_DEVICE_SCANNER",
0x0000001c : "FILE_DEVICE_SCREEN",
0x00000037 : "FILE_DEVICE_SERENUM",
0x0000001a : "FILE_DEVICE_SERIAL_MOUSE_PORT",
0x0000001b : "FILE_DEVICE_SERIAL_PORT",
0x00000031 : "FILE_DEVICE_SMARTCARD",
0x0000002e : "FILE_DEVICE_SMB",
0x0000001d : "FILE_DEVICE_SOUND",
0x0000001e : "FILE_DEVICE_STREAMS",
0x0000001f : "FILE_DEVICE_TAPE",
0x00000020 : "FILE_DEVICE_TAPE_FILE_SYSTEM",
0x00000038 : "FILE_DEVICE_TERMSRV",
0x00000021 : "FILE_DEVICE_TRANSPORT",
0x00000022 : "FILE_DEVICE_UNKNOWN",
0x0000002c : "FILE_DEVICE_VDM",
0x00000023 : "FILE_DEVICE_VIDEO",
0x00000024 : "FILE_DEVICE_VIRTUAL_DISK",
0x00000025 : "FILE_DEVICE_WAVE_IN",
0x00000026 : "FILE_DEVICE_WAVE_OUT",
}
vollog = logging.getLogger(__name__)
class DeviceTree(interfaces.plugins.PluginInterface):
"""Listing tree based on drivers and attached devices in a particular windows memory image."""
_required_framework_version = (2, 0, 3)
_version = (1, 0, 0)
@classmethod
def get_requirements(cls) -> List[interfaces.configuration.RequirementInterface]:
return [
requirements.ModuleRequirement(name = "kernel", description = "Windows kernel",
architectures = ["Intel32", "Intel64"]),
requirements.PluginRequirement(name = "driverscan", plugin = driverscan.DriverScan, version = (1, 0, 0)),
]
def _generator(self) -> Iterator[Tuple]:
kernel = self.context.modules[self.config["kernel"]]
# Scan the Layer for drivers
for driver in driverscan.DriverScan.scan_drivers(self.context, kernel.layer_name, kernel.symbol_table_name):
try:
try:
driver_name = driver.get_driver_name()
except (ValueError, exceptions.PagedInvalidAddressException):
vollog.log(constants.LOGLEVEL_VVVV,
f"Failed to get Driver name : {driver.vol.offset:x}")
driver_name = renderers.UnparsableValue()
yield (0, (
format_hints.Hex(driver.vol.offset),
"DRV",
driver_name,
renderers.NotApplicableValue(),
renderers.NotApplicableValue(),
renderers.NotApplicableValue()
))
# Scan to get the device information of driver.
for device in driver.get_devices():
try:
device_name = device.get_device_name()
except (ValueError, exceptions.PagedInvalidAddressException):
vollog.log(constants.LOGLEVEL_VVVV,
f"Failed to get Device name : {device.vol.offset:x}")
device_name = renderers.UnparsableValue()
device_type = DEVICE_CODES.get(device.DeviceType, "UNKNOWN")
yield (1, (
format_hints.Hex(driver.vol.offset),
"DEV",
driver_name,
device_name,
renderers.NotApplicableValue(),
device_type
))
# Scan to get the attached devices information of device.
for level, attached_device in enumerate(device.get_attached_devices(), start=2):
try:
device_name = attached_device.get_device_name()
except (ValueError, exceptions.PagedInvalidAddressException):
vollog.log(constants.LOGLEVEL_VVVV,
f"Failed to get Attached Device Name: {attached_device.vol.offset:x}")
device_name = renderers.UnparsableValue()
attached_device_driver_name = attached_device.DriverObject.DriverName.get_string()
attached_device_type = DEVICE_CODES.get(attached_device.DeviceType, "UNKNOWN")
yield (level, (
format_hints.Hex(driver.vol.offset),
"ATT",
driver_name,
device_name,
attached_device_driver_name,
attached_device_type
))
except(exceptions.PagedInvalidAddressException):
vollog.log(constants.LOGLEVEL_VVVV,
f"Invalid address identified in drivers and devices: {driver.vol.offset:x}")
continue
def run(self) -> renderers.TreeGrid:
return renderers.TreeGrid([
("Offset", format_hints.Hex),
("Type", str),
("DriverName", str),
("DeviceName", str),
("DriverNameOfAttDevice", str),
("DeviceType", str),
], self._generator())
@@ -4,10 +4,10 @@
import binascii
import hashlib
import logging
from struct import unpack, pack
from typing import List, Tuple, Optional
from struct import pack, unpack
from typing import List, Optional, Tuple
from Crypto.Cipher import ARC4, DES, AES
from Crypto.Cipher import AES, ARC4, DES
from Crypto.Hash import MD5
from volatility3.framework import interfaces, renderers
@@ -28,7 +28,7 @@ class Hashdump(interfaces.plugins.PluginInterface):
def get_requirements(cls):
return [
requirements.ModuleRequirement(name = 'kernel', description = 'Windows kernel',
architectures = ["Intel32", "Intel64"]),
architectures = ["Intel32", "Intel64"]),
requirements.PluginRequirement(name = 'hivelist', plugin = hivelist.HiveList, version = (1, 0, 0))
]
@@ -63,7 +63,8 @@ class Hashdump(interfaces.plugins.PluginInterface):
def get_hive_key(cls, hive: registry.RegistryHive, key: str):
result = None
try:
result = hive.get_key(key)
if hive:
result = hive.get_key(key)
except KeyError:
vollog.info(
f"Unable to load the required registry key {hive.get_name()}\\{key} from this memory image")
@@ -132,7 +133,7 @@ class Hashdump(interfaces.plugins.PluginInterface):
rc4_key = md5.digest()
rc4 = ARC4.new(rc4_key)
hbootkey = rc4.encrypt(sam_data[0x80:0xA0]) # lgtm [py/weak-cryptographic-algorithm]
hbootkey = rc4.encrypt(sam_data[0x80:0xA0]) # lgtm [py/weak-cryptographic-algorithm]
return hbootkey
elif revision == 3:
# AES encrypted
@@ -151,7 +152,7 @@ class Hashdump(interfaces.plugins.PluginInterface):
des2 = DES.new(des_k2, DES.MODE_ECB)
cipher = AES.new(hbootkey[:16], AES.MODE_CBC, salt)
obfkey = cipher.decrypt(enc_hash)
return des1.decrypt(obfkey[:8]) + des2.decrypt(obfkey[8:16]) # lgtm [py/weak-cryptographic-algorithm]
return des1.decrypt(obfkey[:8]) + des2.decrypt(obfkey[8:16]) # lgtm [py/weak-cryptographic-algorithm]
@classmethod
def get_user_hashes(cls, user: registry.CM_KEY_NODE, samhive: registry.RegistryHive,
@@ -229,9 +230,9 @@ class Hashdump(interfaces.plugins.PluginInterface):
md5.update(hbootkey[:0x10] + pack("<L", rid) + lmntstr)
rc4_key = md5.digest()
rc4 = ARC4.new(rc4_key)
obfkey = rc4.encrypt(enc_hash) # lgtm [py/weak-cryptographic-algorithm]
obfkey = rc4.encrypt(enc_hash) # lgtm [py/weak-cryptographic-algorithm]
return des1.decrypt(obfkey[:8]) + des2.decrypt(obfkey[8:]) # lgtm [py/weak-cryptographic-algorithm]
return des1.decrypt(obfkey[:8]) + des2.decrypt(obfkey[8:]) # lgtm [py/weak-cryptographic-algorithm]
@classmethod
def get_user_name(cls, user: registry.CM_KEY_NODE, samhive: registry.RegistryHive) -> Optional[bytes]:
@@ -253,13 +254,9 @@ class Hashdump(interfaces.plugins.PluginInterface):
# replaces the dump_hashes method in vol2
def _generator(self, syshive: registry.RegistryHive, samhive: registry.RegistryHive):
if syshive is None:
vollog.debug("SYSTEM address is None: Did you use the correct profile?")
yield (0, (renderers.NotAvailableValue(), renderers.NotAvailableValue(), renderers.NotAvailableValue(),
renderers.NotAvailableValue()))
vollog.debug("SYSTEM address is None: No system hive found")
if samhive is None:
vollog.debug("SAM address is None: Did you use the correct profile?")
yield (0, (renderers.NotAvailableValue(), renderers.NotAvailableValue(), renderers.NotAvailableValue(),
renderers.NotAvailableValue()))
vollog.debug("SAM address is None: No SAM hive found")
bootkey = self.get_bootkey(syshive)
hbootkey = self.get_hbootkey(samhive, bootkey)
if hbootkey:
@@ -0,0 +1,99 @@
from volatility3.framework import interfaces, constants
from volatility3.framework import renderers, interfaces, exceptions
from volatility3.framework.configuration import requirements
from volatility3.framework.objects import utility
from volatility3.framework.renderers import format_hints
from volatility3.framework.symbols import intermed
from volatility3.framework.symbols.windows.extensions import pe
from volatility3.plugins.windows import pslist, vadinfo
class LdrModules(interfaces.plugins.PluginInterface):
_required_framework_version = (2, 0, 0)
_version = (1, 0, 0)
@classmethod
def get_requirements(cls):
return [
requirements.ModuleRequirement(name = 'kernel', description = 'Windows kernel', architectures = ["Intel32", "Intel64"]),
requirements.VersionRequirement(name = 'pslist', component = pslist.PsList, version = (2, 0, 0)),
requirements.VersionRequirement(name = 'vadinfo', component = vadinfo.VadInfo, version = (2, 0, 0)),
requirements.ListRequirement(name = 'pid',
element_type = int,
description = "Process IDs to include (all other processes are excluded)",
optional = True),
]
def _generator(self, procs):
pe_table_name = intermed.IntermediateSymbolTable.create(self.context,
self.config_path,
"windows",
"pe",
class_types = pe.class_types)
def filter_function(x: interfaces.objects.ObjectInterface) -> bool:
try:
return not (x.get_private_memory() == 0 and x.ControlArea)
except AttributeError:
return False
filter_func = filter_function
for proc in procs:
proc_layer_name = proc.add_process_layer()
# Build dictionaries from different module lists, where the DllBase address is the key and value is the module object
load_order_mod = dict((mod.DllBase, mod)
for mod in proc.load_order_modules())
init_order_mod = dict((mod.DllBase, mod)
for mod in proc.init_order_modules())
mem_order_mod = dict((mod.DllBase, mod)
for mod in proc.mem_order_modules())
# Build dictionary of mapped files, where the VAD start address is the key and value is the file name of the mapped file
mapped_files = {}
for vad in vadinfo.VadInfo.list_vads(proc, filter_func = filter_func):
dos_header = self.context.object(pe_table_name + constants.BANG + "_IMAGE_DOS_HEADER",
offset = vad.get_start(),
layer_name = proc_layer_name)
try:
# Filter out VADs that do not start with a MZ header
if dos_header.e_magic != 0x5A4D:
continue
except exceptions.PagedInvalidAddressException:
continue
mapped_files[vad.get_start()] = vad.get_file_name()
for base in mapped_files.keys():
# Does the base address exist in the PEB DLL lists?
load_mod = load_order_mod.get(base, None)
init_mod = init_order_mod.get(base, None)
mem_mod = mem_order_mod.get(base, None)
yield (0, [int(proc.UniqueProcessId),
str(proc.ImageFileName.cast("string",
max_length = proc.ImageFileName.vol.count,
errors = 'replace')),
format_hints.Hex(base),
load_mod != None,
init_mod != None,
mem_mod != None,
mapped_files[base]])
def run(self):
filter_func = pslist.PsList.create_pid_filter(self.config.get('pid', None))
kernel = self.context.modules[self.config['kernel']]
return renderers.TreeGrid([("Pid", int),
("Process", str),
("Base", format_hints.Hex),
("InLoad", bool),
("InInit", bool),
("InMem", bool),
("MappedPath", str)],
self._generator(
pslist.PsList.list_processes(context = self.context,
layer_name = kernel.layer_name,
symbol_table = kernel.symbol_table_name,
filter_func = filter_func)))
@@ -53,7 +53,7 @@ class Malfind(interfaces.plugins.PluginInterface):
"""
CHUNK_SIZE = 0x1000
all_zero_page = "\x00" * CHUNK_SIZE
all_zero_page = b"\x00" * CHUNK_SIZE
offset = 0
vad_length = vad.get_end() - vad.get_start()
@@ -0,0 +1,200 @@
# This file is Copyright 2022 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 hashlib
from typing import Iterator, List, Tuple
from volatility3.framework import constants, exceptions, interfaces, renderers, symbols
from volatility3.framework.configuration import requirements
from volatility3.framework.layers import scanners
from volatility3.framework.renderers import format_hints
from volatility3.framework.symbols import intermed
from volatility3.framework.symbols.windows.extensions import mbr
vollog = logging.getLogger(__name__)
class MBRScan(interfaces.plugins.PluginInterface):
"""Scans for and parses potential Master Boot Records (MBRs)"""
_required_framework_version = (2, 0, 1)
_version = (1, 0, 0)
@classmethod
def get_requirements(cls)-> List[interfaces.configuration.RequirementInterface]:
return [
requirements.ModuleRequirement(name = 'kernel', description = 'Windows kernel',
architectures = ["Intel32", "Intel64"]),
requirements.BooleanRequirement(name = 'full',
description ="It analyzes and provides all the information in the partition entry and bootcode hexdump. (It returns a lot of information, so we recommend you render it in CSV.)",
default = False,
optional = True)
]
@classmethod
def get_hash(cls, data:bytes) -> str:
return hashlib.md5(data).hexdigest()
def _generator(self) -> Iterator[Tuple]:
kernel = self.context.modules[self.config['kernel']]
physical_layer_name = self.context.layers[kernel.layer_name].config.get('memory_layer', None)
# Decide of Memory Dump Architecture
layer = self.context.layers[physical_layer_name]
architecture = "intel" if not symbols.symbol_table_is_64bit(self.context, kernel.symbol_table_name) else "intel64"
# Read in the Symbol File
symbol_table = intermed.IntermediateSymbolTable.create(context = self.context,
config_path = self.config_path,
sub_path = "windows",
filename = "mbr",
class_types = {
'PARTITION_TABLE': mbr.PARTITION_TABLE,
'PARTITION_ENTRY': mbr.PARTITION_ENTRY
})
partition_table_object = symbol_table + constants.BANG + "PARTITION_TABLE"
# Define Signature and Data Length
mbr_signature = b"\x55\xAA"
mbr_length = 0x200
bootcode_length = 0x1B8
# Scan the Layer for Raw Master Boot Record (MBR) and parse the fields
for offset, _value in layer.scan(context = self.context, scanner = scanners.MultiStringScanner(patterns = [mbr_signature])):
try:
mbr_start_offset = offset - (mbr_length - len(mbr_signature))
partition_table = self.context.object(partition_table_object, offset = mbr_start_offset, layer_name = layer.name)
# Extract only BootCode
full_mbr = layer.read(mbr_start_offset, mbr_length, pad = True)
bootcode = full_mbr[:bootcode_length]
all_zeros = None
if bootcode:
all_zeros = bootcode.count(b"\x00") == len(bootcode)
if not all_zeros:
partition_entries = [
partition_table.FirstEntry, partition_table.SecondEntry,
partition_table.ThirdEntry, partition_table.FourthEntry
]
if not self.config.get("full", True):
yield (0, (
format_hints.Hex(offset),
partition_table.get_disk_signature(),
self.get_hash(bootcode),
self.get_hash(full_mbr),
renderers.NotApplicableValue(),
renderers.NotApplicableValue(),
renderers.NotApplicableValue(),
renderers.NotApplicableValue(),
interfaces.renderers.Disassembly(bootcode, 0, architecture)
))
else:
yield (0, (
format_hints.Hex(offset),
partition_table.get_disk_signature(),
self.get_hash(bootcode),
self.get_hash(full_mbr),
renderers.NotApplicableValue(),
renderers.NotApplicableValue(),
renderers.NotApplicableValue(),
renderers.NotApplicableValue(),
renderers.NotApplicableValue(),
renderers.NotApplicableValue(),
renderers.NotApplicableValue(),
renderers.NotApplicableValue(),
renderers.NotApplicableValue(),
renderers.NotApplicableValue(),
renderers.NotApplicableValue(),
renderers.NotApplicableValue(),
renderers.NotApplicableValue(),
interfaces.renderers.Disassembly(bootcode, 0, architecture),
format_hints.HexBytes(bootcode)
))
for partition_index, partition_entry_object in enumerate(partition_entries, start=1):
if not self.config.get("full", True):
yield (1, (
format_hints.Hex(offset),
partition_table.get_disk_signature(),
self.get_hash(bootcode),
self.get_hash(full_mbr),
partition_index,
partition_entry_object.is_bootable(),
partition_entry_object.get_partition_type(),
format_hints.Hex(partition_entry_object.get_size_in_sectors()),
renderers.NotApplicableValue()
))
else:
yield (1, (
format_hints.Hex(offset),
partition_table.get_disk_signature(),
self.get_hash(bootcode),
self.get_hash(full_mbr),
partition_index,
partition_entry_object.is_bootable(),
format_hints.Hex(partition_entry_object.get_bootable_flag()),
partition_entry_object.get_partition_type(),
format_hints.Hex(partition_entry_object.PartitionType),
format_hints.Hex(partition_entry_object.get_starting_lba()),
partition_entry_object.get_starting_cylinder(),
partition_entry_object.get_starting_chs(),
partition_entry_object.get_starting_sector(),
partition_entry_object.get_ending_cylinder(),
partition_entry_object.get_ending_chs(),
partition_entry_object.get_ending_sector(),
format_hints.Hex(partition_entry_object.get_size_in_sectors()),
renderers.NotApplicableValue(),
renderers.NotApplicableValue()
))
else:
vollog.log(constants.LOGLEVEL_VVVV, f"Not a valid MBR: Data all zeroed out : {format_hints.Hex(offset)}")
continue
except exceptions.PagedInvalidAddressException as excp:
vollog.log(constants.LOGLEVEL_VVVV, f"Invalid address identified in guessed MBR: {hex(excp.invalid_address)}")
continue
def run(self)-> renderers.TreeGrid:
if not self.config.get("full", True):
return renderers.TreeGrid([
("Potential MBR at Physical Offset", format_hints.Hex),
("Disk Signature", str),
("Bootcode MD5", str),
("Full MBR MD5", str),
("PartitionIndex", int),
("Bootable", bool),
("PartitionType", str),
("SectorInSize", format_hints.Hex),
("Disasm", interfaces.renderers.Disassembly)
], self._generator())
else:
return renderers.TreeGrid([
("Potential MBR at Physical Offset", format_hints.Hex),
("Disk Signature", str),
("Bootcode MD5", str),
("Full MBR MD5", str),
("PartitionIndex", int),
("Bootable", bool),
("BootFlag", format_hints.Hex),
("PartitionType", str),
("PartitionTypeRaw", format_hints.Hex),
("StartingLBA", format_hints.Hex),
("StartingCylinder", int),
("StartingCHS", int),
("StartingSector", int),
("EndingCylinder", int),
("EndingCHS", int),
("EndingSector", int),
("SectorInSize", format_hints.Hex),
("Disasm", interfaces.renderers.Disassembly),
("Bootcode", format_hints.HexBytes)
], self._generator())
@@ -0,0 +1,164 @@
# This file is Copyright 2022 Volatility Foundation and licensed under the Volatility Software License 1.0
# which is available at https://www.volatilityfoundation.org/license/vsl-v1.0
#
import datetime
import logging
from volatility3.framework import constants, exceptions, interfaces, renderers
from volatility3.framework.configuration import requirements
from volatility3.framework.renderers import conversion, format_hints
from volatility3.framework.symbols import intermed
from volatility3.framework.symbols.windows.extensions import mft
from volatility3.plugins import timeliner, yarascan
vollog = logging.getLogger(__name__)
class MFTScan(interfaces.plugins.PluginInterface, timeliner.TimeLinerInterface):
"""Scans for MFT FILE objects present in a particular windows memory image."""
_required_framework_version = (2, 0, 0)
@classmethod
def get_requirements(cls):
return [
requirements.TranslationLayerRequirement(name = 'primary',
description = 'Memory layer for the kernel',
architectures = ["Intel32", "Intel64"]),
requirements.VersionRequirement(name = 'yarascanner', component = yarascan.YaraScanner,
version = (2, 0, 0)),
]
def _generator(self):
layer = self.context.layers[self.config['primary']]
# Yara Rule to scan for MFT Header Signatures
rules = yarascan.YaraScan.process_yara_options({'yara_rules': '/FILE0|FILE\*|BAAD/'})
# Read in the Symbol File
symbol_table = intermed.IntermediateSymbolTable.create(context = self.context,
config_path = self.config_path,
sub_path = "windows",
filename = "mft",
class_types = {
'FILE_NAME_ENTRY': mft.MFTFileName,
'MFT_ENTRY': mft.MFTEntry
})
# get each of the individual Field Sets
mft_object = symbol_table + constants.BANG + "MFT_ENTRY"
attribute_object = symbol_table + constants.BANG + "ATTRIBUTE"
header_object = symbol_table + constants.BANG + "ATTR_HEADER"
si_object = symbol_table + constants.BANG + "STANDARD_INFORMATION_ENTRY"
fn_object = symbol_table + constants.BANG + "FILE_NAME_ENTRY"
# Scan the layer for Raw MFT records and parse the fields
for offset, _rule_name, _name, _value in layer.scan(context = self.context,
scanner = yarascan.YaraScanner(rules = rules)):
try:
mft_record = self.context.object(mft_object, offset = offset, layer_name = layer.name)
# We will update this on each pass in the next loop and use it as the new offset.
attr_base_offset = mft_record.FirstAttrOffset
attr_header = self.context.object(header_object,
offset = offset + attr_base_offset,
layer_name = layer.name)
# There is no field that has a count of Attributes
# Keep Attempting to read attributes until we get an invalid attr_header.AttrType
while attr_header.AttrType.is_valid_choice:
vollog.debug(f"Attr Type: {attr_header.AttrType.lookup()}")
# Offset past the headers to the attribute data
attr_data_offset = offset + attr_base_offset + self.context.symbol_space.get_type(
attribute_object).relative_child_offset("Attr_Data")
# MFT Flags determine the file type or dir
# If we don't have a valid enum, coerce to hex so we can keep the record
try:
mft_flag = mft_record.Flags.lookup()
except ValueError:
mft_flag = hex(mft_record.Flags)
# Standard Information Attribute
if attr_header.AttrType.lookup() == 'STANDARD_INFORMATION':
attr_data = self.context.object(si_object, offset = attr_data_offset, layer_name = layer.name)
yield 0, (
format_hints.Hex(attr_data_offset),
mft_record.get_signature(),
mft_record.RecordNumber,
mft_record.LinkCount,
mft_flag,
renderers.NotApplicableValue(),
attr_header.AttrType.lookup(),
conversion.wintime_to_datetime(attr_data.CreationTime),
conversion.wintime_to_datetime(attr_data.ModifiedTime),
conversion.wintime_to_datetime(attr_data.UpdatedTime),
conversion.wintime_to_datetime(attr_data.AccessedTime),
renderers.NotApplicableValue(),
)
# File Name Attribute
if attr_header.AttrType.lookup() == 'FILE_NAME':
attr_data = self.context.object(fn_object, offset = attr_data_offset, layer_name = layer.name)
file_name = attr_data.get_full_name()
# If we don't have a valid enum, coerce to hex so we can keep the record
try:
permissions = attr_data.Flags.lookup()
except ValueError:
permissions = hex(attr_data.Flags)
yield 1, (format_hints.Hex(attr_data_offset), mft_record.get_signature(),
mft_record.RecordNumber, mft_record.LinkCount, mft_flag, permissions,
attr_header.AttrType.lookup(),
conversion.wintime_to_datetime(attr_data.CreationTime),
conversion.wintime_to_datetime(attr_data.ModifiedTime),
conversion.wintime_to_datetime(attr_data.UpdatedTime),
conversion.wintime_to_datetime(attr_data.AccessedTime), file_name)
# If there's no advancement the loop will never end, so break it now
if attr_header.Length == 0:
break
# Update the base offset to point to the next attribute
attr_base_offset += attr_header.Length
# Get the next attribute
attr_header = self.context.object(header_object,
offset = offset + attr_base_offset,
layer_name = layer.name)
except exceptions.PagedInvalidAddressException:
pass
def generate_timeline(self):
for row in self._generator():
_depth, row_data = row
# Only Output FN Records
if row_data[6] == 'FILE_NAME':
filename = row_data[-1]
description = f"MFT FILE_NAME entry for {filename}"
yield (description, timeliner.TimeLinerType.CREATED, row_data[7])
yield (description, timeliner.TimeLinerType.MODIFIED, row_data[8])
yield (description, timeliner.TimeLinerType.CHANGED, row_data[9])
yield (description, timeliner.TimeLinerType.ACCESSED, row_data[10])
def run(self):
return renderers.TreeGrid([
('Offset', format_hints.Hex),
('Record Type', str),
('Record Number', int),
('Link Count', int),
('MFT Type', str),
('Permissions', str),
('Attribute Type', str),
('Created', datetime.datetime),
('Modified', datetime.datetime),
('Updated', datetime.datetime),
('Accessed', datetime.datetime),
('Filename', str),
], self._generator())
@@ -55,14 +55,14 @@ class Privs(interfaces.plugins.PluginInterface):
try:
process_token = task.Token.dereference().cast("_TOKEN")
except exceptions.InvalidAddressException:
vollog.log(constants.LOGLEVEL_VVV, 'Skeep invalid token.')
vollog.log(constants.LOGLEVEL_VVV, 'Skip invalid token.')
continue
for value, present, enabled, default in process_token.privileges():
# Skip privileges whose bit positions cannot be
# translated to a privilege name
if not self.privilege_info.get(int(value)):
vollog.log(constants.LOGLEVEL_VVV, f'Skeep invalid privilege ({value}).')
vollog.log(constants.LOGLEVEL_VVV, f'Skip invalid privilege ({value}).')
continue
name, desc = self.privilege_info.get(int(value))
@@ -85,7 +85,7 @@ class PsScan(interfaces.plugins.PluginInterface, timeliner.TimeLinerInterface):
context: The context to retrieve required elements (layers, symbol tables) from
layer_name: The name of the layer on which to operate
symbol_table: The name of the table containing the kernel symbols
proc: the process object with phisical address
proc: the process object with physical address
Returns:
A process object on virtual address layer
@@ -0,0 +1,103 @@
# This file is Copyright 2022 Volatility Foundation and licensed under the Volatility Software License 1.0
# which is available at https://www.volatilityfoundation.org/license/vsl-v1.0
#
import datetime
import logging
from volatility3.framework import renderers, interfaces
from volatility3.framework.configuration import requirements
from volatility3.framework.objects import utility
from volatility3.plugins.windows import pslist
from volatility3.plugins import timeliner
vollog = logging.getLogger(__name__)
class Sessions(interfaces.plugins.PluginInterface, timeliner.TimeLinerInterface):
"""lists Processes with Session information extracted from Environmental Variables"""
_required_framework_version = (2, 0, 0)
@classmethod
def get_requirements(cls):
return [
requirements.ModuleRequirement(name = 'kernel',
description = 'Windows kernel',
architectures = ["Intel32", "Intel64"]),
requirements.PluginRequirement(name = 'pslist', plugin = pslist.PsList, version = (2, 0, 0)),
requirements.ListRequirement(name = 'pid',
element_type = int,
description = "Process IDs to include (all other processes are excluded)",
optional = True)
]
def _generator(self):
kernel = self.context.modules[self.config['kernel']]
filter_func = pslist.PsList.create_pid_filter(self.config.get('pid', None))
# Collect all the values as we will want to group them later
sessions = {}
for proc in pslist.PsList.list_processes(self.context,
kernel.layer_name,
kernel.symbol_table_name,
filter_func = filter_func):
session_id = proc.get_session_id()
# Detect RDP, Console or set default value
session_type = renderers.NotAvailableValue()
# Construct Username from Process Env
user_domain = ''
user_name = ''
for var, val in proc.environment_variables():
if var.lower() == 'username':
user_name = val
elif var.lower() == 'userdomain':
user_domain = val
if var.lower() == 'sessionname':
session_type = val
# Concat Domain and User
full_user = f'{user_domain}/{user_name}'
if full_user == '/':
full_user = renderers.NotAvailableValue()
# Collect all the values in to a row we can yield after sorting.
row = {
"session_id": session_id,
"process_id": proc.UniqueProcessId,
"process_name": utility.array_to_string(proc.ImageFileName),
"user_name": full_user,
"process_start": proc.get_create_time(),
"session_type": session_type
}
# Add row to correct session so we can sort it later
if session_id in sessions:
sessions[session_id].append(row)
else:
sessions[session_id] = [row]
# Group and yield each row
for rows in sessions.values():
for row in rows:
yield 0, (row.get('session_id'), row.get('session_type'), row.get('process_id'),
row.get('process_name'), row.get('user_name'), row.get('process_start'))
def generate_timeline(self):
for row in self._generator():
_depth, row_data = row
# Only add to timeline if we have the username
# Without the user context PSList output is identical
if isinstance(row_data[4], str):
description = f"Process: {row_data[2]} {row_data[3]} started by user {row_data[4]}"
yield (description, timeliner.TimeLinerType.CREATED, row_data[5])
def run(self):
return renderers.TreeGrid([("Session ID", int), ('Session Type', str), ("Process ID", int), ("Process", str),
("User Name", str), ("Create Time", datetime.datetime)], self._generator())
+11 -3
View File
@@ -3,7 +3,7 @@
#
import logging
from typing import Iterable, Tuple, List, Dict, Any
from typing import Any, Dict, Iterable, List, Tuple
from volatility3.framework import interfaces, renderers
from volatility3.framework.configuration import requirements
@@ -15,8 +15,11 @@ vollog = logging.getLogger(__name__)
try:
import yara
if tuple([int(x) for x in yara.__version__.split('.')]) < (3, 8):
raise ImportError
except ImportError:
vollog.info("Python Yara module not found, plugin (and dependent plugins) not available")
vollog.info("Python Yara (>3.8.0) module not found, plugin (and dependent plugins) not available")
raise
@@ -40,7 +43,10 @@ class YaraScan(plugins.PluginInterface):
"""Scans kernel memory using yara rules (string or file)."""
_required_framework_version = (2, 0, 0)
_version = (1, 0, 0)
_version = (1, 1, 0)
# TODO: When the major version is bumped, take the opportunity to rename the yara_rules config to yara_string
# or something that makes more sense
@classmethod
def get_requirements(cls) -> List[interfaces.configuration.RequirementInterface]:
@@ -83,6 +89,8 @@ class YaraScan(plugins.PluginInterface):
if config.get('wide', False):
rule += " wide ascii"
rules = yara.compile(sources = {'n': f'rule r1 {{strings: $a = {rule} condition: $a}}'})
elif config.get('yara_source', None) is not None:
rules = yara.compile(source = config['yara_source'])
elif config.get('yara_file', None) is not None:
rules = yara.compile(file = resources.ResourceAccessor().open(config['yara_file'], "rb"))
elif config.get('yara_compiled_file', None) is not None:
+13 -7
View File
@@ -272,20 +272,26 @@ class TreeGrid(interfaces.renderers.TreeGrid):
def _append(self, parent: Optional[interfaces.renderers.TreeNode], values: Any) -> TreeNode:
"""Adds a new node at the top level if parent is None, or under the
parent node otherwise, after all other children."""
children = self.children(parent)
return self._insert(parent, len(children), values)
return self._insert(parent, None, values)
def _insert(self, parent: Optional[interfaces.renderers.TreeNode], position: int, values: Any) -> TreeNode:
def _insert(self, parent: Optional[interfaces.renderers.TreeNode], position: Optional[int], values: Any) -> TreeNode:
"""Inserts an element into the tree at a specific position."""
parent_path = ""
children = self._find_children(parent)
if parent is not None:
parent_path = parent.path + self.path_sep
newpath = parent_path + str(position)
if position is None:
newpath = parent_path + str(len(children))
else:
newpath = parent_path + str(position)
for node, _ in children[position:]:
self.visit(node, lambda child, _: child.path_changed(newpath, True), None)
tree_item = TreeNode(newpath, self, parent, values)
for node, _ in children[position:]:
self.visit(node, lambda child, _: child.path_changed(newpath, True), None)
children.insert(position, (tree_item, []))
if position is None:
children.append((tree_item, []))
else:
children.insert(position, (tree_item, []))
return tree_item
def is_ancestor(self, node, descendant):
+7 -3
View File
@@ -11,13 +11,13 @@ import os
import pathlib
import zipfile
from abc import ABCMeta
from typing import Any, Dict, Generator, Iterable, List, Optional, Type, Tuple, Mapping
from typing import Any, Dict, Generator, Iterable, List, Mapping, Optional, Tuple, Type
from volatility3 import schemas, symbols
from volatility3.framework import class_subclasses, constants, exceptions, interfaces, objects
from volatility3.framework.configuration import requirements
from volatility3.framework.layers import resources
from volatility3.framework.symbols import native, metadata
from volatility3.framework.symbols import metadata, native
vollog = logging.getLogger(__name__)
@@ -113,6 +113,9 @@ class IntermediateSymbolTable(interfaces.symbols.SymbolTableInterface):
metadata = json_object.get('metadata', None)
if not metadata:
raise exceptions.SymbolSpaceError(f"Invalid ISF file attempted to be parsed: {isf_url}")
# Determine the delegate or throw an exception
self._delegate = self._closest_version(metadata.get('format', "0.0.0"),
self._versions)(context, config_path, name, json_object, native_types,
@@ -540,7 +543,8 @@ class Version3Format(Version2Format):
if 'type' in symbol:
symbol_type = self._interdict_to_template(symbol['type'])
self._symbol_cache[name] = interfaces.symbols.SymbolInterface(name = name, address = address, type = symbol_type)
self._symbol_cache[name] = interfaces.symbols.SymbolInterface(name = name, address = address,
type = symbol_type)
return self._symbol_cache[name]
@@ -228,6 +228,21 @@ class task_struct(generic.GenericIntelProcess):
"""
return not self.is_kernel_thread and self.tgid != self.pid
def get_threads(self) -> Iterable[interfaces.objects.ObjectInterface]:
"""Returns a list of the task_struct based on the list_head
thread_node structure."""
task_symbol_table_name = self.get_symbol_table_name()
# iterating through the thread_list from thread_group
# this allows iterating through pointers to grab the
# threads and using the thread_group offset to get the
# corresponding task_struct
for task in self.thread_group.to_list(
f"{task_symbol_table_name}{constants.BANG}task_struct",
"thread_group"
):
yield task
class fs_struct(objects.StructType):
@@ -1,7 +1,7 @@
# This file is Copyright 2020 Volatility Foundation and licensed under the Volatility Software License 1.0
# which is available at https://www.volatilityfoundation.org/license/vsl-v1.0
#
import volatility3.framework.symbols.windows.extensions.pool
from volatility3.framework.symbols import intermed
from volatility3.framework.symbols.windows import extensions
from volatility3.framework.symbols.windows.extensions import registry, pool
@@ -7,16 +7,18 @@ import datetime
import functools
import logging
import math
from typing import Iterable, Iterator, Optional, Union, Tuple, List
from typing import Generator, Iterable, Iterator, List, Optional, Tuple, Union
from volatility3.framework import constants, exceptions, interfaces, objects, renderers, symbols
from volatility3.framework.interfaces.objects import ObjectInterface
from volatility3.framework.layers import intel
from volatility3.framework.renderers import conversion
from volatility3.framework.symbols import generic
from volatility3.framework.symbols.windows.extensions import pool, pe, kdbg
from volatility3.framework.symbols.windows.extensions import kdbg, pe, pool
vollog = logging.getLogger(__name__)
# Keep these in a basic module, to prevent import cycles when symbol providers require them
@@ -306,12 +308,15 @@ class MMVAD(MMVAD_SHORT):
try:
# this is for xp and 2003
if self.has_member("ControlArea"):
file_name = self.ControlArea.FilePointer.FileName.get_string()
filename_obj = self.ControlArea.FilePointer.FileName
# this is for vista through windows 7
else:
file_name = self.Subsection.ControlArea.FilePointer.dereference().cast(
"_FILE_OBJECT").FileName.get_string()
filename_obj = self.Subsection.ControlArea.FilePointer.dereference().cast(
"_FILE_OBJECT").FileName
if filename_obj.Length > 0:
file_name = filename_obj.get_string()
except exceptions.InvalidAddressException:
pass
@@ -348,17 +353,32 @@ class DEVICE_OBJECT(objects.StructType, pool.ExecutiveObject):
"""A class for kernel device objects."""
def get_device_name(self) -> str:
"""Get device's name from the object header."""
header = self.get_object_header()
return header.NameInfo.Name.String # type: ignore
def get_attached_devices(self) -> Generator[ObjectInterface, None, None]:
"""Enumerate the attached device's objects"""
device = self.AttachedDevice.dereference()
while device:
yield device
device = device.AttachedDevice.dereference()
class DRIVER_OBJECT(objects.StructType, pool.ExecutiveObject):
"""A class for kernel driver objects."""
def get_driver_name(self) -> str:
"""Get driver's name from the object header."""
header = self.get_object_header()
return header.NameInfo.Name.String # type: ignore
def get_devices(self) -> Generator[ObjectInterface, None, None]:
"""Enumerate the driver's device objects"""
device = self.DeviceObject.dereference()
while device:
yield device
device = device.NextDevice.dereference()
def is_valid(self) -> bool:
"""Determine if the object is valid."""
return True
@@ -461,10 +481,13 @@ class UNICODE_STRING(objects.StructType):
# We explicitly do *not* catch errors here, we allow an exception to be thrown
# (otherwise there's no way to determine anything went wrong)
# It's up to the user of this method to catch exceptions
return self.Buffer.dereference().cast("string",
max_length = self.Length,
errors = "replace",
encoding = "utf16")
# We manually construct an object rather than casting a dereferenced pointer in case
# the buffer length is 0 and the pointer is a NULL pointer
return self._context.object(self.vol.type_name.split(constants.BANG)[0] + constants.BANG + 'string',
layer_name = self.Buffer.vol.layer_name,
offset = self.Buffer,
max_length = self.Length, errors = 'replace', encoding = 'utf16')
String = property(get_string)
@@ -898,8 +921,8 @@ class CONTROL_AREA(objects.StructType):
return False
# The first SubsectionBase should not be page aligned
#subsection = self.get_subsection()
#if subsection.SubsectionBase & self.PAGE_MASK == 0:
# subsection = self.get_subsection()
# if subsection.SubsectionBase & self.PAGE_MASK == 0:
# return False
except exceptions.InvalidAddressException:
return False
@@ -948,7 +971,7 @@ class CONTROL_AREA(objects.StructType):
subsection_offset = starting_sector * 0x200
# Similar to the check in is_valid(), make sure the SubsectionBase is not page aligned.
#if subsection.SubsectionBase & self.PAGE_MASK == 0:
# if subsection.SubsectionBase & self.PAGE_MASK == 0:
# break
ptecount = 0
@@ -979,8 +1002,8 @@ class CONTROL_AREA(objects.StructType):
# Currently just a temporary workaround to deal with custom bit flag
# in the PFN field for pages in transition state.
# See https://github.com/volatilityfoundation/volatility3/pull/475
physoffset = (mmpte.u.Trans.PageFrameNumber & (( 1 << 33 ) - 1 ) ) << 12
physoffset = (mmpte.u.Trans.PageFrameNumber & ((1 << 33) - 1)) << 12
yield physoffset, file_offset, self.PAGE_SIZE
# Go to the next PTE entry
@@ -0,0 +1,62 @@
# This file is Copyright 2022 Volatility Foundation and licensed under the Volatility Software License 1.0
# which is available at https://www.volatilityfoundation.org/license/vsl-v1.0
#
from volatility3.framework import objects
class PARTITION_TABLE(objects.StructType):
def get_disk_signature(self) -> str:
"""Get Disk Signature (GUID)."""
return "{0:02x}-{1:02x}-{2:02x}-{3:02x}".format(
self.DiskSignature[0],
self.DiskSignature[1],
self.DiskSignature[2],
self.DiskSignature[3]
)
class PARTITION_ENTRY(objects.StructType):
def get_bootable_flag(self) -> int:
"""Get Bootable Flag."""
return self.BootableFlag
def is_bootable(self) -> bool:
"""Check Bootable Partition."""
return False if not (self.get_bootable_flag() == 0x80) else True
def get_partition_type(self) -> str:
"""Get Partition Type."""
return self.PartitionType.lookup() if self.PartitionType.is_valid_choice else "Not Defined PartitionType"
def get_starting_chs(self) -> int:
"""Get Starting CHS (Cylinder Header Sector) Address."""
return self.StartingCHS[0]
def get_ending_chs(self) -> int:
"""Get Ending CHS (Cylinder Header Sector) Address."""
return self.EndingCHS[0]
def get_starting_sector(self) -> int:
"""Get Starting Sector."""
return self.StartingCHS[1] % 64
def get_ending_sector(self) -> int:
"""Get Ending Sector."""
return self.EndingCHS[1] % 64
def get_starting_cylinder(self) -> int:
"""Get Starting Cylinder."""
return (self.StartingCHS[1] - self.get_starting_sector()) * 4 + self.StartingCHS[2]
def get_ending_cylinder(self) -> int:
"""Get Ending Cylinder."""
return (self.EndingCHS[1] - self.get_ending_sector()) * 4 + self.EndingCHS[2]
def get_starting_lba(self) -> int:
"""Get Starting LBA (Logical Block Addressing)."""
return self.StartingLBA
def get_size_in_sectors(self) -> int:
"""Get Size in Sectors."""
return self.SizeInSectors
@@ -0,0 +1,21 @@
# This file is Copyright 2022 Volatility Foundation and licensed under the Volatility Software License 1.0
# which is available at https://www.volatilityfoundation.org/license/vsl-v1.0
#
from volatility3.framework import objects
class MFTEntry(objects.StructType):
"""This represents the base MFT Record"""
def get_signature(self) -> str:
signature = self.Signature.cast('string', max_length = 4, encoding = 'latin-1')
return signature
class MFTFileName(objects.StructType):
"""This represents an MFT $FILE_NAME Attribute"""
def get_full_name(self) -> str:
output = self.Name.cast("string", encoding = "utf16", max_length = self.NameLength * 2, errors = "replace")
return output
@@ -2,15 +2,15 @@
# which is available at https://www.volatilityfoundation.org/license/vsl-v1.0
#
from typing import Generator, Tuple
import logging
from typing import Generator, Tuple
from volatility3.framework import constants
from volatility3.framework import objects, interfaces
from volatility3.framework import constants, interfaces, objects
from volatility3.framework.renderers import conversion
vollog = logging.getLogger(__name__)
class IMAGE_DOS_HEADER(objects.StructType):
def get_nt_header(self) -> interfaces.objects.ObjectInterface:
@@ -77,12 +77,13 @@ class IMAGE_DOS_HEADER(objects.StructType):
image_base_type = nt_header.OptionalHeader.ImageBase.vol.type_name
member_size = self._context.symbol_space.get_type(image_base_type).size
try:
newval = objects.convert_value_to_data(self.vol.offset, int, nt_header.OptionalHeader.ImageBase.vol.data_format)
newval = objects.convert_value_to_data(self.vol.offset, int,
nt_header.OptionalHeader.ImageBase.vol.data_format)
new_pe = raw_data[:image_base_offset] + newval + raw_data[image_base_offset + member_size:]
except OverflowError:
vollog.warning("Volatility was unable to fix the image base for the PE file at base address {:#x}. " \
"This will cause issues with many static analysis tools if you do not inform the " \
"tool of the in-memory load address.".format(self.vol.offset))
"This will cause issues with many static analysis tools if you do not inform the " \
"tool of the in-memory load address.".format(self.vol.offset))
new_pe = raw_data
return new_pe
@@ -109,7 +110,7 @@ class IMAGE_DOS_HEADER(objects.StructType):
size_of_image = nt_header.OptionalHeader.SizeOfImage
# no legitimate PE is going to be larger than this
if size_of_image > (1024 * 1024 * 100):
if size_of_image > constants.windows.PE_MAX_EXTRACTION_SIZE:
raise ValueError(f"The claimed SizeOfImage is too large: {size_of_image}")
read_layer = self._context.layers[layer_name]
@@ -128,8 +128,8 @@ class POOL_HEADER(objects.StructType):
# ---------------
if addr - optional_headers_length < 0:
continue
padding_length = struct.unpack(
"<I", infomask_data[addr - optional_headers_length:addr - optional_headers_length + 4])[0]
padding_length, = struct.unpack(
"<I", infomask_data[addr - optional_headers_length:addr - optional_headers_length + 4])
padding_length -= lengths_of_optional_headers[padding_available or 0]
# Certain versions of windows have PADDING_INFO lengths that are too long
@@ -233,7 +233,10 @@ class POOL_TRACKER_BIG_PAGES(objects.StructType):
def is_valid(self) -> bool:
return self.Key > 0
# return self.Va > 0x1
def is_free(self) -> bool:
"""Returns if the allocation is freed (True) or in-use (False)"""
return self.Va & 1 == 1
def get_key(self) -> str:
"""Returns the Key value as a 4 character string"""
@@ -5,10 +5,10 @@
import enum
import logging
import struct
from typing import Optional, Iterable, Union
from typing import Iterable, Optional, Union
from volatility3.framework import constants, exceptions, objects, interfaces
from volatility3.framework.layers.registry import RegistryHive, RegistryInvalidIndex, RegistryFormatException
from volatility3.framework import constants, exceptions, interfaces, objects
from volatility3.framework.layers.registry import RegistryFormatException, RegistryHive, RegistryInvalidIndex
vollog = logging.getLogger(__name__)
@@ -76,7 +76,9 @@ class CMHIVE(objects.StructType):
for attr in ["FileFullPath", "FileUserName", "HiveRootPath"]:
try:
return getattr(self, attr).get_string()
name = getattr(self, attr)
if name.Length > 0:
return name.get_string()
except (AttributeError, exceptions.InvalidAddressException):
pass
@@ -264,19 +266,22 @@ class CM_KEY_VALUE(objects.StructType):
if self_type == RegValueTypes.REG_DWORD:
if len(data) != struct.calcsize("<L"):
raise ValueError(f"Size of data does not match the type of registry value {self.get_name()}")
return struct.unpack("<L", data)[0]
res, = struct.unpack("<L", data)
return res
if self_type == RegValueTypes.REG_DWORD_BIG_ENDIAN:
if len(data) != struct.calcsize(">L"):
raise ValueError(f"Size of data does not match the type of registry value {self.get_name()}")
return struct.unpack(">L", data)[0]
res, = struct.unpack(">L", data)
return res
if self_type == RegValueTypes.REG_QWORD:
if len(data) != struct.calcsize("<Q"):
raise ValueError(f"Size of data does not match the type of registry value {self.get_name()}")
return struct.unpack("<Q", data)[0]
res, = struct.unpack("<Q", data)
return res
if self_type in [
RegValueTypes.REG_SZ, RegValueTypes.REG_EXPAND_SZ, RegValueTypes.REG_LINK, RegValueTypes.REG_MULTI_SZ,
RegValueTypes.REG_BINARY, RegValueTypes.REG_FULL_RESOURCE_DESCRIPTOR, RegValueTypes.REG_RESOURCE_LIST,
RegValueTypes.REG_RESOURCE_REQUIREMENTS_LIST
RegValueTypes.REG_SZ, RegValueTypes.REG_EXPAND_SZ, RegValueTypes.REG_LINK, RegValueTypes.REG_MULTI_SZ,
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:
@@ -0,0 +1,240 @@
{
"metadata": {
"producer": {
"version": "0.0.1",
"name": "Donghyun Kim (@digitalisx99)",
"comment": "Using structures defined in File System Forensic Analysis pg 88+",
"datetime": "2022-03-05T10:53:00"
},
"format": "6.1.0"
},
"base_types": {
"unsigned long": {
"kind": "int",
"size": 4,
"signed": false,
"endian": "little"
},
"unsigned long long": {
"kind": "int",
"size": 8,
"signed": false,
"endian": "little"
},
"long": {
"kind": "int",
"size": 4,
"signed": true,
"endian": "little"
},
"unsigned int": {
"kind": "int",
"size": 4,
"signed": false,
"endian": "little"
},
"int": {
"kind": "int",
"size": 4,
"signed": true,
"endian": "little"
},
"unsigned short": {
"kind": "int",
"size": 2,
"signed": false,
"endian": "little"
},
"unsigned char": {
"kind": "int",
"size": 1,
"signed": false,
"endian": "little"
},
"wchar": {
"kind": "int",
"size": 2,
"signed": true,
"endian": "little"
}
},
"symbols": {},
"enums": {
"PartitionTypes": {
"base": "unsigned char",
"constants": {
"Empty": 0,
"FAT12,CHS": 1,
"FAT16 16-32MB,CHS": 4,
"Microsoft Extended": 5,
"FAT16 32MB,CHS": 6,
"NTFS": 7,
"FAT32,CHS": 11,
"FAT32,LBA": 12,
"FAT16, 32MB-2GB,LBA": 14,
"Microsoft Extended, LBA": 15,
"Hidden FAT12,CHS": 17,
"Hidden FAT16,16-32MB,CHS": 20,
"Hidden FAT16,32MB-2GB,CHS": 22,
"AST SmartSleep Partition": 24,
"Hidden FAT32,CHS": 27,
"Hidden FAT32,LBA": 28,
"Hidden FAT16,32MB-2GB,LBA": 30,
"PQservice": 39,
"Plan 9 partition": 57,
"PartitionMagic recovery partition": 60,
"Microsoft MBR,Dynamic Disk": 66,
"GoBack partition": 68,
"Novell": 81,
"CP/M": 82,
"Unix System V": 99,
"PC-ARMOUR protected partition": 100,
"Solaris x86 or Linux Swap": 130,
"Linux": 131,
"Hibernation": 132,
"Linux Extended": 133,
"NTFS Volume Set": 134,
"NTFS Volume Set": 135,
"BSD/OS": 159,
"Hibernation": 160,
"Hibernation": 161,
"FreeBSD": 165,
"OpenBSD": 166,
"Mac OSX": 168,
"NetBSD": 169,
"Mac OSX Boot": 171,
"MacOS X HFS": 175,
"BSDI": 183,
"BSDI Swap": 184,
"Boot Wizard hidden": 187,
"Solaris 8 boot partition": 190,
"CP/M-86": 216,
"Dell PowerEdge Server utilities (FAT fs)": 222,
"DG/UX virtual disk manager partition": 223,
"BeOS BFS": 235,
"EFI GPT Disk": 238,
"EFI System Partition": 239,
"VMWare File System": 251,
"VMWare Swap": 252
},
"size": 1
}
},
"user_types": {
"PARTITION_ENTRY":{
"fields": {
"BootableFlag": {
"offset": 0,
"type": {
"kind": "base",
"name": "unsigned char"
}
},
"StartingCHS": {
"offset": 1,
"type": {
"count": 3,
"kind": "array",
"subtype": {
"kind": "base",
"name": "unsigned char"
}
}
},
"PartitionType": {
"offset": 4,
"type": {
"kind": "enum",
"name": "PartitionTypes"
}
},
"EndingCHS": {
"offset": 5,
"type": {
"count": 3,
"kind": "array",
"subtype": {
"kind": "base",
"name": "unsigned char"
}
}
},
"StartingLBA": {
"offset": 8,
"type": {
"kind": "base",
"name": "unsigned int"
}
},
"SizeInSectors": {
"offset": 12,
"type": {
"kind": "base",
"name": "unsigned int"
}
}
},
"kind": "struct",
"size": 16
},
"PARTITION_TABLE":{
"fields":{
"DiskSignature": {
"offset": 440,
"type": {
"count": 4,
"kind": "array",
"subtype": {
"kind": "base",
"name": "unsigned char"
}
}
},
"Unused": {
"offset": 444,
"type": {
"kind": "base",
"name": "unsigned short"
}
},
"FirstEntry":{
"offset": 446,
"type": {
"kind": "struct",
"name": "PARTITION_ENTRY"
}
},
"SecondEntry":{
"offset": 462,
"type": {
"kind": "struct",
"name": "PARTITION_ENTRY"
}
},
"ThirdEntry":{
"offset": 478,
"type": {
"kind": "struct",
"name": "PARTITION_ENTRY"
}
},
"FourthEntry":{
"offset": 494,
"type": {
"kind": "struct",
"name": "PARTITION_ENTRY"
}
},
"Signature":{
"offset": 510,
"type": {
"kind": "base",
"name": "unsigned short"
}
}
},
"kind": "struct",
"size": 512
}
}
}
@@ -0,0 +1,469 @@
{
"metadata": {
"producer": {
"version": "0.0.1",
"name": "kevthehermit-by-hand",
"comment": "Using structures defined in File System Forensic Analysis pg 353+",
"datetime": "2022-01-03T13:37:00"
},
"format": "6.1.0"
},
"base_types": {
"unsigned long": {
"kind": "int",
"size": 4,
"signed": false,
"endian": "little"
},
"unsigned long long": {
"kind": "int",
"size": 8,
"signed": false,
"endian": "little"
},
"long": {
"kind": "int",
"size": 4,
"signed": true,
"endian": "little"
},
"unsigned int": {
"kind": "int",
"size": 4,
"signed": false,
"endian": "little"
},
"unsigned short": {
"kind": "int",
"size": 2,
"signed": false,
"endian": "little"
},
"unsigned char": {
"kind": "int",
"size": 1,
"signed": false,
"endian": "little"
},
"wchar": {
"kind": "int",
"size": 2,
"signed": true,
"endian": "little"
}
},
"symbols": {},
"enums": {
"AttrTypeEnum": {
"base": "unsigned char",
"constants": {
"STANDARD_INFORMATION": 16,
"ATTRIBUTE_LIST": 32,
"FILE_NAME": 48,
"OBJECT_ID": 64,
"SECURITY_DESCRIPTOR": 80,
"VOLUME_NAME": 96,
"VOLUME_INFORMATION": 112,
"DATA": 128,
"INDEX_ROOT": 114,
"INDEX_ALLOCATION": 160,
"BITMAP": 176,
"REPARSE_POINT": 192,
"EA_INFORMATION": 208,
"EA": 224,
"PROPERTY_SET": 240,
"LOGGED_UTILITY_STREAM": 256
},
"size": 1
},
"NameSpaceEnum": {
"base":"unsigned char",
"constants": {
"POSIX": 0,
"Win32": 1,
"DOS": 2,
"Win32 DOS": 3
},
"size": 1
},
"MFTFlagsEnum": {
"base":"unsigned char",
"constants": {
"Removed": 0,
"File": 1,
"Directory": 2,
"DirInUse": 3
},
"size": 1
},
"PermissionFlagEnum": {
"base":"unsigned char",
"constants": {
"ReadOnly": 1,
"Hidden": 2,
"System": 4,
"Archive": 32,
"ArchiveHidden": 34,
"ArchiveSystem": 36,
"ArchiveHiddenSystem": 38,
"Device": 60,
"Normal": 128,
"Temporary": 256,
"TempArchive": 288,
"SparseFile": 512,
"ReparsePoint": 1024,
"Compressed": 2048,
"Offline": 4096,
"NotIndexed": 8192,
"Encrypted": 16384,
"Directory": 268435456,
"IndexView": 536870912
},
"size": 1
}
},
"user_types": {
"MFT_ENTRY": {
"fields": {
"Signature": {
"offset": 0,
"type": {
"count": 1,
"kind": "array",
"subtype": {
"kind": "base",
"name": "unsigned char"
}
}
},
"UpdateSequenceOffset": {
"offset": 4,
"type": {
"kind": "base",
"name": "unsigned short"
}
},
"NumFixupEntries": {
"offset": 6,
"type": {
"kind": "base",
"name": "unsigned short"
}
},
"LSN": {
"offset": 8,
"type": {
"kind": "base",
"name": "unsigned long long"
}
},
"SequenceValue": {
"offset": 16,
"type": {
"kind": "base",
"name": "unsigned short"
}
},
"LinkCount": {
"offset": 18,
"type": {
"kind": "base",
"name": "unsigned short"
}
},
"FirstAttrOffset": {
"offset": 20,
"type":{
"kind": "base",
"name": "unsigned short"
}
},
"Flags": {
"offset": 22,
"type":{
"kind": "enum",
"name": "MFTFlagsEnum"
}
},
"RealSize": {
"offset": 24,
"type":{
"kind": "base",
"name": "unsigned int"
}
},
"AllocatedSize": {
"offset": 28,
"type":{
"kind": "base",
"name": "unsigned int"
}
},
"BaseReference": {
"offset": 32,
"type":{
"kind": "base",
"name": "unsigned long long"
}
},
"NextAttrID": {
"offset": 40,
"type":{
"kind": "base",
"name": "unsigned short"
}
},
"RecordNumber": {
"offset": 44,
"type":{
"kind": "base",
"name": "unsigned long"
}
}
},
"kind": "struct",
"size": 1024
},
"ATTRIBUTE": {
"fields":{
"Attr_Header": {
"offset": 0,
"type": {
"kind": "struct",
"name": "mft!ATTR_HEADER"
}
},
"Resident_Header": {
"offset": 16,
"type": {
"kind": "struct",
"name": "mft!RESIDENT_HEADER"
}
},
"Attr_Data": {
"offset": 24,
"type": {
"kind": "struct",
"name": "mft!ATTR_HEADER"
}
}
},
"kind": "struct",
"size": 96
},
"ATTR_HEADER": {
"fields": {
"AttrType": {
"offset": 0,
"type": {
"kind": "enum",
"name": "AttrTypeEnum"
}
},"Length": {
"offset": 4,
"type": {
"kind": "base",
"name": "unsigned int"
}
},
"NonResidentFlag": {
"offset": 8,
"type": {
"kind": "base",
"name": "unsigned char"
}
},
"NameLength": {
"offset": 9,
"type": {
"kind": "base",
"name": "unsigned char"
}
},
"NameOffset": {
"offset": 10,
"type": {
"kind": "base",
"name": "unsigned short"
}
},
"Flags": {
"offset": 12,
"type": {
"kind": "enum",
"name": "MFTFlagsEnum"
}
},
"AttributeID": {
"offset": 14,
"type": {
"kind": "base",
"name": "unsigned short"
}
}
},
"kind": "struct",
"size": 16
},"RESIDENT_HEADER": {
"fields": {
"AttrSize": {
"offset": 0,
"type": {
"kind": "base",
"name": "unsigned int"
}
},"AttrOffset": {
"offset": 4,
"type": {
"kind": "base",
"name": "unsigned int"
}
},
"IndexFlag": {
"offset": 8,
"type": {
"kind": "base",
"name": "unsigned short"
}
}
},
"kind": "struct",
"size": 8
},
"STANDARD_INFORMATION_ENTRY": {
"fields": {
"CreationTime": {
"offset": 0,
"type": {
"kind": "base",
"name": "unsigned long long"
}
},
"ModifiedTime": {
"offset": 8,
"type": {
"kind": "base",
"name": "unsigned long long"
}
},
"UpdatedTime": {
"offset": 16,
"type": {
"kind": "base",
"name": "unsigned long long"
}
},
"AccessedTime": {
"offset": 24,
"type": {
"kind": "base",
"name": "unsigned long long"
}
},
"flags": {
"offset": 32,
"type": {
"kind": "enum",
"name": "PermissionFlagEnum"
}
}
},
"kind": "struct",
"size": 1024
},
"FILE_NAME_ENTRY": {
"fields": {
"ParentDirectory": {
"offset": 0,
"type": {
"kind": "base",
"name": "unsigned long long"
}
},
"CreationTime": {
"offset": 8,
"type": {
"kind": "base",
"name": "unsigned long long"
}
},
"ModifiedTime": {
"offset": 16,
"type": {
"kind": "base",
"name": "unsigned long long"
}
},
"UpdatedTime": {
"offset": 24,
"type": {
"kind": "base",
"name": "unsigned long long"
}
},
"AccessedTime": {
"offset": 32,
"type": {
"kind": "base",
"name": "unsigned long long"
}
},
"AllocatedFileSize": {
"offset": 40,
"type": {
"kind": "base",
"name": "unsigned long long"
}
},
"RealFileSize": {
"offset": 48,
"type": {
"kind": "base",
"name": "unsigned long long"
}
},
"Flags": {
"offset": 56,
"type": {
"kind": "enum",
"name": "PermissionFlagEnum"
}
},
"ReparseValue": {
"offset": 60,
"type": {
"kind": "base",
"name": "unsigned int"
}
},
"NameLength": {
"offset": 64,
"type": {
"kind": "base",
"name": "unsigned char"
}
},
"NameSpace": {
"offset": 65,
"type": {
"kind": "base",
"name": "unsigned char"
}
},
"Name": {
"offset": 66,
"type": {
"count": 10,
"kind": "array",
"subtype": {
"kind": "base",
"name": "wchar"
}
}
}
},
"kind": "struct",
"size": 1024
}
}
}
@@ -131,16 +131,22 @@ class PDBUtility(interfaces.configuration.VersionableInterface):
# Check it is actually the MZ header
if mz_sig != b"MZ":
return None
nt_header_start = ord(layer.read(offset + 0x3C, 1))
optional_header_size = struct.unpack('<H', layer.read(offset + nt_header_start + 0x14, 2))[0]
nt_header_start, = struct.unpack("<I", layer.read(offset + 0x3C, 4))
pe_sig = layer.read(offset + nt_header_start, 2)
# Check it is actually the Nt Headers
if pe_sig != b"PE":
return None
optional_header_size, = struct.unpack('<H', layer.read(offset + nt_header_start + 0x14, 2))
# Just enough to tell us the max size
pe_header = layer.read(offset, nt_header_start + 0x16 + optional_header_size)
pe_data = pefile.PE(data = pe_header)
max_size = pe_data.OPTIONAL_HEADER.SizeOfImage
# Proper data
virtual_data = layer.read(offset, max_size)
virtual_data = layer.read(offset, max_size, pad=True)
pe_data = pefile.PE(data = virtual_data)
# De-virtualize the memory
@@ -357,4 +363,4 @@ class PdbSignatureScanner(interfaces.layers.ScannerInterface):
guid = (16 * '{:02X}').format(g0, g1, g2, g3, g4, g5, g6, g7, g8, g9, ga, gb, gc, gd, ge, gf)
if match.start(0) < self.chunk_size:
yield (guid, a, pdb_name, match.start(0))
yield (guid, a, pdb_name, data_offset + match.start(0))