WEBDIENSTE/volatility3
1
0
Fork 0
mirror of https://github.com/volatilityfoundation/volatility3.git synced 2026-09-06 17:57:38 +02:00
Code Issues Packages Projects Releases Wiki Activity

Merge branch 'develop' into volshell_display_types_pointer_upgrade_2025

Browse Source
This commit is contained in:
Eve
2025-04-03 16:54:28 +01:00
committed by GitHub
parent 86945492a7 ed1b1f5369
commit 61c67b63d4
76 changed files with 1499 additions and 425 deletions
Download Patch File Download Diff File Expand all files Collapse all files
.github/workflows/vol3-code-analysis.yml
+24
View File
@@ -0,0 +1,24 @@
name: Volatility3 Code Analysis
on: [push, pull_request]
jobs:
build:
runs-on: ubuntu-22.04
strategy:
matrix:
python-version: ["3.8"]
steps:
- uses: actions/checkout@v4
- name: Set up Python ${{ matrix.python-version }}
uses: actions/setup-python@v5
with:
python-version: ${{ matrix.python-version }}
- name: Install dependencies
run: |
python -m pip install --upgrade pip
pip install .[test]
- name: Testing...
run: |
python ./test/volatility3_code_analysis.py
API_CHANGES.md
+94 -2
View File
@@ -4,6 +4,100 @@ 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.25.0
======
Pointer class now supports `get_raw_value()`.
`KTIMER` no longer supports `get_raw_dpc()`.
2.24.0
======
Support `encoding` parameter for `objects.utility.array_to_string`
2.23.0
======
Add support for windows GUI classes and OS distinguishers.
Add a symbol_table_name for `ExecutiveObject.get_object_header()`/
2.22.0
======
Linux net constants added.
Network objects moved to separate versionable module.
2.21.0
======
`uuid` method added to `linux.extensions`.
2.20.0
======
NM_TYPES_DESC constants added to linux.
`latch_tree_root` and `kernel_symbol` added to linux extensions.
Linux `module` class additions:
* `get_module_address_boundaries`
* `section_typetab`
Linux `task_struct` class additions:
* `get_address_space_layer`
* `state`
Linux `bpf_prog` class additions:
* `bpf_jit_binary_hdr_address`
2.19.0
======
Introduction of `Modules` versionable linux extension module.
Deprecation of some `LinuxUtilities` functions relating to modules.
2.18.0
======
Addition of `scatterlist` linux extension.
2.17.0
======
The addition of a `types` member to `SymbolInterface`
2.16.0
======
Addition of TAINT_FLAG constants, `TaintFlag` dataclass
Addition of linux `tainting` versionable module
2.15.0
======
Addition of `convert_fourcc_code` to `LinuxUtilities` class
2.14.0
======
No significant changes (part of the 2.16.0 PR which took time in development)
2.13.0
======
Linux `task` object extension addition of `getppid`
2.12.0
======
Changes to the Intel layer to support `PROT_NONE` pages.
2.11.0
======
Addition of `get_type` method to windows `CM_KEY_NODE` registry structure
2.10.0
======
No significant API changes (CLI changes to the JSONL text renderer)
2.9.0
=====
No significant API changes (change to call `linux.LinuxUtilities.get_module_from_volobj_type` to get the kernel)
2.8.0
=====
Addition of the `BinOrAbsent`, `HexOrAbsent`, `HexBytesOrAbsent` and `MultiTypeDataOrAbsent` data type renderers
2.7.0
=====
Addition of `is_valid`, `get_create_time` and `get_exit_time` to ETHREAD structure
2.6.0
=====
No significant changes (again, the version got bump twice in the PR straight to 2.7.0)
2.5.0
=====
Add in support for specifying a type override for object_from_symbol
@@ -50,5 +144,3 @@ an absolute offset. This can be done with `Module.get_absolute_symbol_address`
* Added context.modules
* Added ModuleRequirement
* Added get\_symbols\_by\_absolute\_location
test/plugins/windows/test_data/windows.registry.userassist.UserAssist.json
+9 -9
View File
@@ -9,7 +9,7 @@
"Last Write Time": "2025-03-06T17:57:09+00:00",
"Name": null,
"Path": "ntuser.dat\\SOFTWARE\\Microsoft\\Windows\\CurrentVersion\\Explorer\\UserAssist\\{F4E57C4B-2036-45F0-A9AB-443BCFE33D9F}\\Count",
"Raw Data": "",
"Raw Data": "N/A",
"Time Focused": null,
"Type": "Key",
"__children": [
@@ -23,7 +23,7 @@
"Last Write Time": "2025-03-06T17:57:09+00:00",
"Name": "%ALLUSERSPROFILE%\\Microsoft\\Windows\\Start Menu\\Programs\\Accessories\\Paint.lnk",
"Path": "ntuser.dat\\SOFTWARE\\Microsoft\\Windows\\CurrentVersion\\Explorer\\UserAssist\\{F4E57C4B-2036-45F0-A9AB-443BCFE33D9F}\\Count",
"Raw Data": "\"\n00 00 00 00 07 00 00 00 00 00 00 00 07 00 00 00 ................\n00 00 80 bf 00 00 80 bf 00 00 80 bf 00 00 80 bf ................\n00 00 80 bf 00 00 80 bf 00 00 80 bf 00 00 80 bf ................\n00 00 80 bf 00 00 80 bf ff ff ff ff 90 86 6b 31 ..............k1\nfd 8d db 01 00 00 00 00 ........ \"",
"Raw Data": "00 00 00 00 07 00 00 00 00 00 00 00 07 00 00 00 00 00 80 bf 00 00 80 bf 00 00 80 bf 00 00 80 bf 00 00 80 bf 00 00 80 bf 00 00 80 bf 00 00 80 bf 00 00 80 bf 00 00 80 bf ff ff ff ff 90 86 6b 31 fd 8d db 01 00 00 00 00",
"Time Focused": "0:00:00.507000",
"Type": "Value",
"__children": []
@@ -38,7 +38,7 @@
"Last Write Time": "2025-03-06T17:57:09+00:00",
"Name": "%ALLUSERSPROFILE%\\Microsoft\\Windows\\Start Menu\\Programs\\Administrative Tools\\Registry Editor.lnk",
"Path": "ntuser.dat\\SOFTWARE\\Microsoft\\Windows\\CurrentVersion\\Explorer\\UserAssist\\{F4E57C4B-2036-45F0-A9AB-443BCFE33D9F}\\Count",
"Raw Data": "\"\n00 00 00 00 01 00 00 00 00 00 00 00 01 00 00 00 ................\n00 00 80 bf 00 00 80 bf 00 00 80 bf 00 00 80 bf ................\n00 00 80 bf 00 00 80 bf 00 00 80 bf 00 00 80 bf ................\n00 00 80 bf 00 00 80 bf ff ff ff ff f0 82 cf ca ................\n95 8e db 01 00 00 00 00 ........ \"",
"Raw Data": "00 00 00 00 01 00 00 00 00 00 00 00 01 00 00 00 00 00 80 bf 00 00 80 bf 00 00 80 bf 00 00 80 bf 00 00 80 bf 00 00 80 bf 00 00 80 bf 00 00 80 bf 00 00 80 bf 00 00 80 bf ff ff ff ff f0 82 cf ca 95 8e db 01 00 00 00 00",
"Time Focused": "0:00:00.501000",
"Type": "Value",
"__children": []
@@ -53,7 +53,7 @@
"Last Write Time": "2025-03-06T17:57:09+00:00",
"Name": "%APPDATA%\\Microsoft\\Windows\\Start Menu\\Programs\\Windows PowerShell\\Windows PowerShell.lnk",
"Path": "ntuser.dat\\SOFTWARE\\Microsoft\\Windows\\CurrentVersion\\Explorer\\UserAssist\\{F4E57C4B-2036-45F0-A9AB-443BCFE33D9F}\\Count",
"Raw Data": "\"\n00 00 00 00 04 00 00 00 00 00 00 00 04 00 00 00 ................\n00 00 80 bf 00 00 80 bf 00 00 80 bf 00 00 80 bf ................\n00 00 80 bf 00 00 80 bf 00 00 80 bf 00 00 80 bf ................\n00 00 80 bf 00 00 80 bf ff ff ff ff 10 67 cf 4d .............g.M\nbe 8e db 01 00 00 00 00 ........ \"",
"Raw Data": "00 00 00 00 04 00 00 00 00 00 00 00 04 00 00 00 00 00 80 bf 00 00 80 bf 00 00 80 bf 00 00 80 bf 00 00 80 bf 00 00 80 bf 00 00 80 bf 00 00 80 bf 00 00 80 bf 00 00 80 bf ff ff ff ff 10 67 cf 4d be 8e db 01 00 00 00 00",
"Time Focused": "0:00:00.504000",
"Type": "Value",
"__children": []
@@ -68,7 +68,7 @@
"Last Write Time": "2025-03-06T17:57:09+00:00",
"Name": "%APPDATA%\\Microsoft\\Windows\\Start Menu\\Programs\\System Tools\\Command Prompt.lnk",
"Path": "ntuser.dat\\SOFTWARE\\Microsoft\\Windows\\CurrentVersion\\Explorer\\UserAssist\\{F4E57C4B-2036-45F0-A9AB-443BCFE33D9F}\\Count",
"Raw Data": "\"\n00 00 00 00 01 00 00 00 00 00 00 00 01 00 00 00 ................\n00 00 80 bf 00 00 80 bf 00 00 80 bf 00 00 80 bf ................\n00 00 80 bf 00 00 80 bf 00 00 80 bf 00 00 80 bf ................\n00 00 80 bf 00 00 80 bf ff ff ff ff d0 99 66 6c ..............fl\nc0 8e db 01 00 00 00 00 ........ \"",
"Raw Data": "00 00 00 00 01 00 00 00 00 00 00 00 01 00 00 00 00 00 80 bf 00 00 80 bf 00 00 80 bf 00 00 80 bf 00 00 80 bf 00 00 80 bf 00 00 80 bf 00 00 80 bf 00 00 80 bf 00 00 80 bf ff ff ff ff d0 99 66 6c c0 8e db 01 00 00 00 00",
"Time Focused": "0:00:00.501000",
"Type": "Value",
"__children": []
@@ -83,7 +83,7 @@
"Last Write Time": "2025-03-06T17:57:09+00:00",
"Name": "%ALLUSERSPROFILE%\\Microsoft\\Windows\\Start Menu\\Programs\\Accessories\\Notepad.lnk",
"Path": "ntuser.dat\\SOFTWARE\\Microsoft\\Windows\\CurrentVersion\\Explorer\\UserAssist\\{F4E57C4B-2036-45F0-A9AB-443BCFE33D9F}\\Count",
"Raw Data": "\"\n00 00 00 00 01 00 00 00 00 00 00 00 01 00 00 00 ................\n00 00 80 bf 00 00 80 bf 00 00 80 bf 00 00 80 bf ................\n00 00 80 bf 00 00 80 bf 00 00 80 bf 00 00 80 bf ................\n00 00 80 bf 00 00 80 bf ff ff ff ff 00 62 ba 89 .............b..\nc0 8e db 01 00 00 00 00 ........ \"",
"Raw Data": "00 00 00 00 01 00 00 00 00 00 00 00 01 00 00 00 00 00 80 bf 00 00 80 bf 00 00 80 bf 00 00 80 bf 00 00 80 bf 00 00 80 bf 00 00 80 bf 00 00 80 bf 00 00 80 bf 00 00 80 bf ff ff ff ff 00 62 ba 89 c0 8e db 01 00 00 00 00",
"Time Focused": "0:00:00.501000",
"Type": "Value",
"__children": []
@@ -98,7 +98,7 @@
"Last Write Time": "2025-03-06T17:57:09+00:00",
"Name": "%ALLUSERSPROFILE%\\Microsoft\\Windows\\Start Menu\\Programs\\Administrative Tools\\Task Scheduler.lnk",
"Path": "ntuser.dat\\SOFTWARE\\Microsoft\\Windows\\CurrentVersion\\Explorer\\UserAssist\\{F4E57C4B-2036-45F0-A9AB-443BCFE33D9F}\\Count",
"Raw Data": "\"\n00 00 00 00 02 00 00 00 00 00 00 00 02 00 00 00 ................\n00 00 80 bf 00 00 80 bf 00 00 80 bf 00 00 80 bf ................\n00 00 80 bf 00 00 80 bf 00 00 80 bf 00 00 80 bf ................\n00 00 80 bf 00 00 80 bf ff ff ff ff b0 24 49 23 .............$I#\nc1 8e db 01 00 00 00 00 ........ \"",
"Raw Data": "00 00 00 00 02 00 00 00 00 00 00 00 02 00 00 00 00 00 80 bf 00 00 80 bf 00 00 80 bf 00 00 80 bf 00 00 80 bf 00 00 80 bf 00 00 80 bf 00 00 80 bf 00 00 80 bf 00 00 80 bf ff ff ff ff b0 24 49 23 c1 8e db 01 00 00 00 00",
"Time Focused": "0:00:00.502000",
"Type": "Value",
"__children": []
@@ -113,11 +113,11 @@
"Last Write Time": "2025-03-06T17:57:09+00:00",
"Name": "%ALLUSERSPROFILE%\\Microsoft\\Windows\\Start Menu\\Programs\\Microsoft Edge.lnk",
"Path": "ntuser.dat\\SOFTWARE\\Microsoft\\Windows\\CurrentVersion\\Explorer\\UserAssist\\{F4E57C4B-2036-45F0-A9AB-443BCFE33D9F}\\Count",
"Raw Data": "\"\n00 00 00 00 01 00 00 00 00 00 00 00 01 00 00 00 ................\n00 00 80 bf 00 00 80 bf 00 00 80 bf 00 00 80 bf ................\n00 00 80 bf 00 00 80 bf 00 00 80 bf 00 00 80 bf ................\n00 00 80 bf 00 00 80 bf ff ff ff ff 60 3d 89 2e ............`=..\nc1 8e db 01 00 00 00 00 ........ \"",
"Raw Data": "00 00 00 00 01 00 00 00 00 00 00 00 01 00 00 00 00 00 80 bf 00 00 80 bf 00 00 80 bf 00 00 80 bf 00 00 80 bf 00 00 80 bf 00 00 80 bf 00 00 80 bf 00 00 80 bf 00 00 80 bf ff ff ff ff 60 3d 89 2e c1 8e db 01 00 00 00 00",
"Time Focused": "0:00:00.501000",
"Type": "Value",
"__children": []
}
]
}
}
}
test/plugins/windows/windows.py
+20 -17
View File
@@ -1,10 +1,10 @@
import json
import hashlib
import shutil
import contextlib
import tempfile
import hashlib
import json
import os
from test import test_volatility, WindowsSamples
import shutil
import tempfile
from test import WindowsSamples, test_volatility
class TestWindowsVolshell:
@@ -843,20 +843,22 @@ class TestWindowsMFTscan:
{
"ADS Filename": "Zone.Identifier",
"Filename": "libby_hoeler_part1.wmv",
"Hexdump": '"\n5b 5a 6f 6e 65 54 72 61 6e 73 66 65 72 5d 0d 0a [ZoneTransfer]..\n5a 6f 6e 65 49 64 3d 33 0d 0a ZoneId=3.. "',
"Hexdump": "5b 5a 6f 6e 65 54 72 61 6e 73 66 65 72 5d 0d 0a 5a 6f 6e 65 49 64 3d 33 0d 0a",
"MFT Type": "DATA",
"Offset": 55926304,
"Record Number": 323,
"Record Type": "FILE",
"__children": [],
},
{
"ADS Filename": "Zone.Identifier",
"Filename": "NetZeroQuickHelpLite.exe",
"Hexdump": '"\n5b 5a 6f 6e 65 54 72 61 6e 73 66 65 72 5d 0d 0a [ZoneTransfer]..\n5a 6f 6e 65 49 64 3d 33 0d 0a ZoneId=3.. "',
"Hexdump": "5b 5a 6f 6e 65 54 72 61 6e 73 66 65 72 5d 0d 0a 5a 6f 6e 65 49 64 3d 33 0d 0a",
"MFT Type": "DATA",
"Offset": 56102400,
"Record Number": 347,
"Record Type": "FILE",
"__children": [],
},
]
for expected_row in expected_rows:
@@ -877,20 +879,22 @@ class TestWindowsMFTscan:
{
"ADS Filename": "$Max",
"Filename": "$UsnJrnl",
"Hexdump": '"\n00 00 00 02 00 00 00 00 00 00 80 00 00 00 00 00 ................\nb9 dd f0 cc df 73 db 01 00 00 00 00 00 00 00 00 .....s.........."',
"Hexdump": "00 00 00 02 00 00 00 00 00 00 80 00 00 00 00 00 b9 dd f0 cc df 73 db 01 00 00 00 00 00 00 00 00",
"MFT Type": "DATA",
"Offset": 1058018088,
"Offset": 26235616,
"Record Number": 107240,
"Record Type": "FILE",
"__children": [],
},
{
"ADS Filename": "$Config",
"Filename": "$Repair",
"Hexdump": '"\n01 00 00 00 03 00 00 00 ........ "',
"ADS Filename": "$SRAT",
"Filename": "$Bitmap",
"Hexdump": "a4 5f fd 60 38 00 01 03 10 00 0c 00 04 00 00 00 01 00 00 00 01 00 00 00 8d 4e 16 00 02 00 00 00 a0 00 00 00 00 00 06 00 03 00 00 00 01 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 4a 7b 01 00 00 00 00 00",
"MFT Type": "DATA",
"Offset": 5009678688,
"Record Number": 28,
"Offset": 1052277088,
"Record Number": 6,
"Record Type": "FILE",
"__children": [],
},
]
for expected_row in expected_rows:
@@ -924,7 +928,7 @@ class TestWindowsMFTscan:
expected_rows = [
{
"Filename": "index",
"Hexdump": '"\n30 5c 72 a7 1b 6d fb fc 09 00 00 00 00 00 00 00 0\\r..m..........\n00 00 00 00 00 00 00 00 ........ "',
"Hexdump": "30 5c 72 a7 1b 6d fb fc 09 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00",
"MFT Type": "DATA",
"Offset": 4961536280,
"Record Number": 116474,
@@ -932,7 +936,7 @@ class TestWindowsMFTscan:
},
{
"Filename": "0.2.filtertrie.intermediate.txt",
"Hexdump": '"\n30 09 32 0d 0a 0.2.. "',
"Hexdump": "30 09 32 0d 0a",
"MFT Type": "DATA",
"Offset": 619242944,
"Record Number": 113013,
@@ -1411,4 +1415,3 @@ class TestWindowsVirtMap:
)
for expected_row in expected_rows:
assert test_volatility.match_output_row(expected_row, json_out)
test/volatility3_code_analysis.py
+440
View File
@@ -0,0 +1,440 @@
"""
This script performs syntax analysis on the volatility3 source tree through a combination of AST analysis and import-time introspection of classes.
The current checks it implements are:
1. Ensure that classes derived from `ConfigurableInterface` properly
declare all `VersionableInterface` classes that they make use of in their
`get_requirements()` classmethod.
:WARNING: a notable exception to this are classes defined within factory
functions. Because these classes are not created until the factory function
is called, they therefore do no exist at import time and cannot be checked
by this script. It is important to keep in mind during code review that
this is a best-effort check and does not make guarantees about the
completeness of declared requirements.
"""
import abc
import argparse
import ast
import importlib
import inspect
import logging
import pkgutil
import sys
import traceback
import types
from typing import Any, Iterator, List, Optional, Tuple, Type, Union
from volatility3.framework import configuration, interfaces
from volatility3.framework.deprecation import PluginRenameClass
logging.basicConfig(format="%(levelname)s: %(message)s")
logger = logging.getLogger(__name__)
class NodeVisitor:
def visit(self, node):
"""Visit a node."""
method = "visit_" + node.__class__.__name__
visitor = getattr(self, method, self.generic_visit)
self.enter(node)
result = visitor(node)
self.leave(node)
return result
def enter(self, node):
"""Called when entering a node."""
method = "enter_" + node.__class__.__name__
visitor = getattr(self, method, self.generic_enter)
return visitor(node)
def leave(self, node):
"""Called when leaving a node."""
method = "leave_" + node.__class__.__name__
visitor = getattr(self, method, self.generic_leave)
return visitor(node)
def generic_visit(self, node):
"""Called if no explicit visitor function exists for a node."""
for _, value in ast.iter_fields(node):
if isinstance(value, list):
for item in value:
if isinstance(item, ast.AST):
self.visit(item)
elif isinstance(value, ast.AST):
self.visit(value)
def generic_enter(self, node):
"""Default enter behavior."""
def generic_leave(self, node):
"""Default leave behavior."""
class CodeViolation(metaclass=abc.ABCMeta):
def __init__(self, module: types.ModuleType, node: ast.AST) -> None:
self.module = module
self.node = node
def __str__(self):
return f"Issue in module {self.module.__name__}: line {self.node.lineno}, col {self.node.col_offset}"
class UnrequiredVersionableUsage(CodeViolation):
def __init__(
self,
module: types.ModuleType,
node: ast.AST,
consuming_class: str,
versionable_item_class: str,
) -> None:
super().__init__(module, node)
self.consuming_class = consuming_class
self.versionable_item_class = versionable_item_class
def __str__(self) -> str:
return (
super().__str__()
+ ": "
+ (
f"Found usage of {self.versionable_item_class} "
f"in class {self.consuming_class} that is not declared "
f"in {self.consuming_class}'s `get_requirements()` classmethod"
)
)
class DirectVolatilityImportUsage(CodeViolation):
def __init__(
self,
module: types.ModuleType,
node: ast.AST,
importing_module: str,
imported_item: object,
imported_name: str,
) -> None:
self.imported_item = imported_item
self.imported_name = imported_name
self.importing_module = importing_module
super().__init__(module, node)
def __str__(self) -> str:
components = self.importing_module.split(".")
return (
super().__str__()
+ ": "
+ (
f"Direct import of {self.imported_name} "
f"({type(self.imported_item)}) "
f"from module {self.importing_module} - "
"change to "
f"'from {'.'.join(components[:-1])} import {components[-1]} and using {components[-1]}.{self.imported_name}"
)
)
def is_versionable(var):
try:
return (
issubclass(var, interfaces.configuration.VersionableInterface)
and var is not interfaces.configuration.VersionableInterface
and not inspect.isabstract(var)
and not (hasattr(var, "hidden") and getattr(var, "hidden") is True)
)
except TypeError:
return False
def is_configurable(var):
try:
return issubclass(var, interfaces.configuration.ConfigurableInterface)
except TypeError:
return False
class ModuleVisitor(NodeVisitor):
def __init__(self, module: types.ModuleType) -> None:
self._module = module
self._scopes = []
self._violations = []
@property
def violations(self):
return self._violations
def _check_vol3_import_from(self, node: ast.ImportFrom):
"""
Ensure that the only thing imported from a volatility3 module (apart
from the root volatility3 module) are functions and modules. This
prevents re-exporting of classes and variables from modules that use
them.
"""
if (
node.module
and node.module.startswith("volatility3.") # Give a pass to volatility3 module
and node.module != "volatility3.framework.constants._version" # make an exception for this
):
for name in node.names:
try:
item = vars(self._module)[
name.asname if name.asname is not None else name.name
]
except KeyError:
logger.debug(
"Couldn't find imported name %s in module %s",
name.asname or name.name,
self._module.__name__,
)
continue
if not (isinstance(item, types.ModuleType) or inspect.isfunction(item)):
self._violations.append(
DirectVolatilityImportUsage(
self._module,
node,
node.module,
item,
name.asname or name.name,
)
)
def enter_ImportFrom(self, node: ast.ImportFrom):
self._check_vol3_import_from(node)
def enter_ClassDef(self, node: ast.ClassDef) -> Any:
logger.debug("Entering class %s", node.name)
clazz = None
try:
clazz = vars(self._module)[str(node.name)]
except KeyError:
logger.debug(
"Failed to get %s from module scope: (%s)",
node.name,
self._module.__name__,
)
if self._scopes:
try:
logger.debug(
"Attempting to get class %s from scope of %s",
node.name,
self._scopes[-1].__name__,
)
clazz = getattr(self._scopes[-1], node.name)
except AttributeError:
logger.debug(
"Class not found in scope of %s", self._scopes[-1].__name__
)
if clazz:
self._scopes.append(clazz)
if clazz and is_configurable(clazz):
logger.info("Checking configurable class %s", clazz.__name__)
visitor = ConfigurableClassVisitor(self._module, clazz)
visitor.visit(node)
self._violations += visitor.violations
self.generic_visit(node)
def leave_ClassDef(self, node: ast.ClassDef):
logger.debug("Leaving class %s", node.name)
try:
scoped_class = next(
scope for scope in self._scopes if scope.__name__ == node.name
)
self._scopes.remove(scoped_class)
except StopIteration:
logger.debug("%s not found in scope list", node.name)
class ConfigurableClassVisitor(NodeVisitor):
def __init__(
self,
module: types.ModuleType,
clazz: Optional[Type[interfaces.configuration.ConfigurableInterface]],
) -> None:
self._module = module
self._current_object = None
self._clazz = clazz
self._seen = set()
self._violations: List[CodeViolation] = []
@property
def versioned_classes(self):
return (
[
req._component
for req in self._clazz.get_requirements()
if isinstance(req, configuration.requirements.VersionRequirement)
]
if self._clazz is not None
else []
)
def check_item(self, item: Type, node: Union[ast.Name, ast.Attribute]):
if (
is_versionable(item)
and self._clazz is not None
and item not in self.versioned_classes
and item is not self._clazz
and not issubclass(self._clazz, PluginRenameClass)
):
logger.info(
"Found versionable item %s, checking against %s",
str(item),
str(self.versioned_classes),
)
result = UnrequiredVersionableUsage(
self._module, node, self._clazz.__name__, item.__name__
)
self._violations.append(result)
@property
def violations(self):
return self._violations
def visit_Name(self, node: ast.Name):
try:
logger.debug(
"Checking module %s for name %s", self._module.__name__, node.id
)
item = vars(self._module)[str(node.id)]
logger.debug("Found %s in %s namespace", node.id, self._module.__name__)
except KeyError:
return
self.check_item(item, node)
def visit_Attribute(
self, node: ast.Attribute
) -> Optional[UnrequiredVersionableUsage]:
if self._clazz is None:
self.generic_visit(node)
return
if (node.lineno, node.col_offset) in self._seen:
return
self._seen.add((node.lineno, node.col_offset))
stack = []
root = node
while True:
stack.append(node.attr)
if isinstance(node.value, ast.Attribute):
node = node.value
elif isinstance(node.value, ast.Name):
stack.append(node.value.id)
break
else:
break
current = None
logger.debug("Checking %s", ".".join(stack[::-1]))
for item in stack[::-1]:
try:
current = (
vars(self._module)[item]
if current is None
else getattr(current, item)
)
except (KeyError, AttributeError) as exc:
logger.debug(
"Failed to get attribute %s (%s)%s",
item,
exc.__class__.__name__,
(" on" + str(current)) if current is not None else "",
)
break
self.check_item(current, root)
def report_missing_requirements() -> Iterator[Tuple[str, UnrequiredVersionableUsage]]:
vol3 = importlib.import_module("volatility3")
for _, module_name, _ in pkgutil.walk_packages(
vol3.__path__, vol3.__name__ + ".", onerror=lambda _: None
):
modname = module_name.replace(
"volatility3.framework.plugins", "volatility3.plugins"
)
try:
# import the module that we want to check
plugin_module = importlib.import_module(modname)
except ImportError as exc:
logger.warning("Failed to import %s: %s", modname, str(exc))
continue
except Exception as exc:
logger.warning(
"An unexpected exception occurred while importing %s: %s",
modname,
str(exc),
)
traceback.print_exc()
continue
logger.info("Checking module %s", plugin_module.__name__)
if plugin_module.__file__ is None:
logger.warning("Plugin module %s has no source file", modname)
continue
try:
with open(plugin_module.__file__, "rb") as f:
source = f.read()
except OSError:
logger.warning(
"Failed to read file contents for %s", plugin_module.__file__
)
continue
try:
module_ast_root = ast.parse(source)
except (SyntaxError, ValueError) as exc:
logger.warning(
"Failed to parse source for %s: %s", plugin_module.__file__, str(exc)
)
raise
mod_visitor = ModuleVisitor(plugin_module)
mod_visitor.visit(module_ast_root)
if mod_visitor.violations:
yield from (
(plugin_module.__name__, res) for res in iter(mod_visitor.violations)
)
def perform_review():
found = 0
for mod, usage in report_missing_requirements():
found += 1
print(str(usage))
if found:
print(f"Found {found} issues")
sys.exit(1)
print("All configurable classes passed validation!")
def parse_args() -> argparse.Namespace:
parser = argparse.ArgumentParser()
parser.add_argument("-v", "--verbose", action="count", dest="verbosity", default=0)
return parser.parse_args()
if __name__ == "__main__":
args = parse_args()
if args.verbosity == 0:
logger.setLevel(logging.WARNING)
elif args.verbosity == 1:
logger.setLevel(logging.INFO)
elif args.verbosity > 1:
logger.setLevel(logging.DEBUG)
perform_review()
volatility3/cli/text_renderer.py
+132 -35
View File
@@ -9,7 +9,7 @@ import random
import string
import sys
from functools import wraps
from typing import Any, Callable, Dict, List, Tuple
from typing import Any, Callable, Dict, List, Optional, Set, Tuple, TypeVar, Union
from volatility3.cli import text_filter
from volatility3.framework import exceptions, interfaces, renderers
@@ -80,7 +80,12 @@ def multitypedata_as_text(value: format_hints.MultiTypeData) -> str:
return hex_bytes_as_text(value)
def optional(func: Callable) -> Callable:
T = TypeVar("T")
def optional(
func: Callable[[Union[interfaces.renderers.BaseAbsentValue, T]], str],
) -> Callable[[T], str]:
@wraps(func)
def wrapped(x: Any) -> str:
if isinstance(x, interfaces.renderers.BaseAbsentValue):
@@ -110,7 +115,7 @@ def quoted_optional(func: Callable) -> Callable:
return wrapped
def display_disassembly(disasm: interfaces.renderers.Disassembly) -> str:
def display_disassembly(disasm: renderers.Disassembly) -> str:
"""Renders a disassembly renderer type into string format.
Args:
@@ -137,13 +142,116 @@ def display_disassembly(disasm: interfaces.renderers.Disassembly) -> str:
return QuickTextRenderer._type_renderers[bytes](disasm.data)
class CLITypeRenderer(interfaces.renderers.TypeRendererInterface):
def __init__(self, func):
super().__init__(func=optional(func))
class LayerDataRenderer(CLITypeRenderer):
"""Renders a LayerData object into data/bytes"""
def __init__(self):
self.context_byte_len = 0
self.width = 16
self.display_offset = False
self.display_hex = True
self.display_ascii = True
def render(
data: Union[renderers.LayerData, interfaces.renderers.BaseAbsentValue],
) -> str:
if isinstance(data, interfaces.renderers.BaseAbsentValue):
# FIXME: Do something cleverer here
return ""
specific_data, error_bytes = self.render_bytes(data)
printables = ""
output = "\n"
for count, byte in enumerate(specific_data):
if count not in error_bytes:
output += f"{byte:02x} "
char = chr(byte)
printables += char if 0x20 <= byte <= 0x7E else "."
else:
output += "__ "
printables += "."
if count % self.width == self.width - 1:
output += printables
if count < len(specific_data) - 1:
output += "\n"
printables = ""
# Handle leftovers when the length is not mutiple of width
if printables:
padding = self.width - len(printables)
output += " " * padding
output += printables
output += " " * padding
return output
render_func = render
return super().__init__(render_func)
def render_bytes(self, data: renderers.LayerData) -> Tuple[bytes, Set[int]]:
"""Renders a valid LayerData into bytes (with context bytes)"""
context_byte_len = self.context_byte_len if not data.no_surrounding else 0
layer = data.context.layers[data.layer_name]
# Map of the holes
error_bytes = set()
start_offset = data.offset - context_byte_len
end_offset = data.offset + data.length + context_byte_len
if isinstance(layer, interfaces.layers.TranslationLayerInterface):
error_bytes = set()
mapping = iter(layer.mapping(start_offset, end_offset, True))
current_map = next(mapping)
for i in range(start_offset, end_offset):
# Run through the bytes, check if they're present
offset, sublength, _, _, _ = current_map
if i < offset:
error_bytes.add(i - start_offset)
if i > offset + sublength:
try:
current_map = next(mapping)
except StopIteration:
pass
offset, sublength, _, _, _ = current_map
if i > offset + sublength:
error_bytes.add(i - start_offset)
# Padded data
specific_data = data.context.layers[data.layer_name].read(
start_offset,
end_offset - start_offset,
True,
)
return specific_data, error_bytes
class CLIRenderer(interfaces.renderers.Renderer):
"""Class to add specific requirements for CLI renderers."""
_type_renderers = {
format_hints.Bin: CLITypeRenderer(lambda x: f"0b{x:b}"),
format_hints.Hex: CLITypeRenderer(lambda x: f"0x{x:x}"),
format_hints.HexBytes: CLITypeRenderer(hex_bytes_as_text),
format_hints.MultiTypeData: CLITypeRenderer(multitypedata_as_text),
renderers.Disassembly: CLITypeRenderer(display_disassembly),
bytes: CLITypeRenderer(lambda x: " ".join(f"{b:02x}" for b in x)),
renderers.LayerData: LayerDataRenderer(),
datetime.datetime: CLITypeRenderer(
lambda x: x.strftime("%Y-%m-%d %H:%M:%S.%f %Z")
),
"default": CLITypeRenderer(lambda x: f"{x}"),
}
name = "unnamed"
structured_output = False
filter: text_filter.CLIFilter = None
column_hide_list: list = None
filter: Optional[text_filter.CLIFilter] = None
column_hide_list: Optional[list] = None
def ignored_columns(
self,
@@ -170,21 +278,11 @@ class CLIRenderer(interfaces.renderers.Renderer):
class QuickTextRenderer(CLIRenderer):
_type_renderers = {
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: quoted_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 = "quick"
def get_render_options(self):
pass
return []
def render(self, grid: interfaces.renderers.TreeGrid) -> None:
"""Renders each column immediately to stdout.
@@ -242,7 +340,7 @@ class NoneRenderer(CLIRenderer):
name = "none"
def get_render_options(self):
pass
return []
def render(self, grid: interfaces.renderers.TreeGrid) -> None:
if not grid.populated:
@@ -250,22 +348,12 @@ class NoneRenderer(CLIRenderer):
class CSVRenderer(CLIRenderer):
_type_renderers = {
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"
structured_output = True
def get_render_options(self):
pass
return []
def render(self, grid: interfaces.renderers.TreeGrid) -> None:
"""Renders each row immediately to stdout.
@@ -316,12 +404,10 @@ class CSVRenderer(CLIRenderer):
class PrettyTextRenderer(CLIRenderer):
_type_renderers = QuickTextRenderer._type_renderers
name = "pretty"
def get_render_options(self):
pass
return []
def render(self, grid: interfaces.renderers.TreeGrid) -> None:
"""Renders each column immediately to stdout.
@@ -380,7 +466,9 @@ class PrettyTextRenderer(CLIRenderer):
accumulator.append((node.path_depth, line))
return accumulator
final_output: List[Tuple[int, Dict[interfaces.renderers.Column, bytes]]] = []
final_output: List[Tuple[int, Dict[interfaces.renderers.Column, list[str]]]] = (
[]
)
if not grid.populated:
grid.populate(visitor, final_output)
else:
@@ -447,9 +535,18 @@ class PrettyTextRenderer(CLIRenderer):
class JsonRenderer(CLIRenderer):
_type_renderers = {
format_hints.HexBytes: quoted_optional(hex_bytes_as_text),
interfaces.renderers.Disassembly: quoted_optional(display_disassembly),
format_hints.HexBytes: lambda x: (
x.hex(" ")
if not isinstance(x, interfaces.renderers.BaseAbsentValue)
else "N/A"
),
renderers.Disassembly: quoted_optional(display_disassembly),
format_hints.MultiTypeData: quoted_optional(multitypedata_as_text),
renderers.LayerData: lambda x: (
LayerDataRenderer().render_bytes(x)[0].hex(" ")
if not isinstance(x, interfaces.renderers.BaseAbsentValue)
else "N/A"
),
bytes: optional(lambda x: " ".join(f"{b:02x}" for b in x)),
datetime.datetime: lambda x: (
x.isoformat()
@@ -463,7 +560,7 @@ class JsonRenderer(CLIRenderer):
structured_output = True
def get_render_options(self) -> List[interfaces.renderers.RenderOption]:
pass
return []
def output_result(self, outfd, result):
"""Outputs the JSON data to a file in a particular format"""
volatility3/cli/volshell/generic.py
+18 -7
View File
@@ -40,6 +40,8 @@ class Volshell(interfaces.plugins.PluginInterface):
_required_framework_version = (2, 0, 0)
_version = (1, 0, 0)
DEFAULT_NUM_DISPLAY_BYTES = 128
def __init__(self, *args, **kwargs):
@@ -54,9 +56,18 @@ class Volshell(interfaces.plugins.PluginInterface):
@classmethod
def get_requirements(cls) -> List[interfaces.configuration.RequirementInterface]:
reqs: List[interfaces.configuration.RequirementInterface] = []
reqs: List[interfaces.configuration.RequirementInterface] = [
requirements.VersionRequirement(
name="regex_scanner",
component=scanners.RegExScanner,
version=(1, 0, 0),
),
]
if cls == Volshell:
reqs = [
reqs += [
requirements.TranslationLayerRequirement(
name="primary", description="Memory layer for the kernel"
),
requirements.URIRequirement(
name="script",
description="File to load and execute at start",
@@ -70,11 +81,8 @@ class Volshell(interfaces.plugins.PluginInterface):
optional=True,
),
]
return reqs + [
requirements.TranslationLayerRequirement(
name="primary", description="Memory layer for the kernel"
),
]
return reqs
def run(
self, additional_locals: Dict[str, Any] = {}
@@ -502,6 +510,7 @@ class Volshell(interfaces.plugins.PluginInterface):
relative_offset, member_type = volobject.vol.members[member]
len_offset = len(hex(relative_offset))
len_member = len(member)
member_type_name = self._get_type_name_with_pointer(
member_type
) # special case for pointers to show what they point to
@@ -509,6 +518,7 @@ class Volshell(interfaces.plugins.PluginInterface):
if len(member_type_name) > MAX_TYPENAME_DISPLAY_LENGTH:
len_typename = MAX_TYPENAME_DISPLAY_LENGTH
member_type_name = f"{member_type_name[:len_typename - 3]}..."
if isinstance(volobject, interfaces.objects.ObjectInterface):
# We're an instance, so also display the data
try:
@@ -593,6 +603,7 @@ class Volshell(interfaces.plugins.PluginInterface):
return f"{display_type_string} @ {hex(volobject.vol.offset)} -> {self._display_value(volobject)}"
else:
return f"{display_type_string}: {self._display_value(volobject)}"
else:
return display_type_string
volatility3/cli/volshell/linux.py
+6 -1
View File
@@ -36,7 +36,12 @@ class Volshell(generic.Volshell):
requirements.IntRequirement(
name="pid", description="Process ID", optional=True
),
]
requirements.VersionRequirement(
name="generic_volshell",
component=generic.Volshell,
version=(1, 0, 0),
),
] + super().get_requirements()
def change_task(self, pid=None):
"""Change the current process and layer, based on a process ID"""
volatility3/cli/volshell/mac.py
+6 -1
View File
@@ -25,7 +25,12 @@ class Volshell(generic.Volshell):
requirements.IntRequirement(
name="pid", description="Process ID", optional=True
),
]
requirements.VersionRequirement(
name="generic_volshell",
component=generic.Volshell,
version=(1, 0, 0),
),
] + super().get_requirements()
def change_task(self, pid=None):
"""Change the current process and layer, based on a process ID"""
volatility3/cli/volshell/windows.py
+6 -1
View File
@@ -23,7 +23,12 @@ class Volshell(generic.Volshell):
requirements.IntRequirement(
name="pid", description="Process ID", optional=True
),
]
requirements.VersionRequirement(
name="generic_volshell",
component=generic.Volshell,
version=(1, 0, 0),
),
] + super().get_requirements()
def change_process(self, pid=None):
"""Change the current process and layer, based on a process ID"""
volatility3/framework/automagic/pdbscan.py
+19 -4
View File
@@ -17,7 +17,7 @@ from volatility3.framework import constants, exceptions, interfaces, layers
from volatility3.framework.configuration import requirements
from volatility3.framework.layers import intel, scanners
from volatility3.framework.symbols import native
from volatility3.framework.symbols.windows.pdbutil import PDBUtility
from volatility3.framework.symbols.windows import pdbutil
if __name__ == "__main__":
import sys
@@ -50,6 +50,21 @@ class KernelPDBScanner(interfaces.automagic.AutomagicInterface):
max_pdb_size = 0x400000
exclusion_list = ["linux", "mac"]
@classmethod
def get_requirements(cls) -> List[interfaces.configuration.RequirementInterface]:
return [
requirements.VersionRequirement(
name="pdb_utility",
component=pdbutil.PDBUtility,
version=(1, 0, 1),
),
requirements.VersionRequirement(
name="bytes_scanner",
component=scanners.BytesScanner,
version=(1, 0, 0),
),
]
def find_virtual_layers_from_req(
self,
context: interfaces.context.ContextInterface,
@@ -120,7 +135,7 @@ class KernelPDBScanner(interfaces.automagic.AutomagicInterface):
):
raise TypeError("PDB name or GUID not a string value")
PDBUtility.load_windows_symbol_table(
pdbutil.PDBUtility.load_windows_symbol_table(
context=context,
guid=kernel["GUID"],
age=kernel["age"],
@@ -259,7 +274,7 @@ class KernelPDBScanner(interfaces.automagic.AutomagicInterface):
bytes(name + ".pdb", "utf-8")
for name in constants.windows.KERNEL_MODULE_NAMES
]
kernels = PDBUtility.pdbname_scan(
kernels = pdbutil.PDBUtility.pdbname_scan(
ctx=context,
layer_name=layer_to_scan,
start=start_scan_address,
@@ -362,7 +377,7 @@ class KernelPDBScanner(interfaces.automagic.AutomagicInterface):
with contextlib.suppress(exceptions.InvalidAddressException):
if vlayer.read(address, 0x2) == b"MZ":
res = list(
PDBUtility.pdbname_scan(
pdbutil.PDBUtility.pdbname_scan(
ctx=context,
layer_name=vlayer.name,
page_size=vlayer.page_size,
volatility3/framework/automagic/symbol_finder.py
+6 -1
View File
@@ -40,7 +40,12 @@ class SymbolFinder(interfaces.automagic.AutomagicInterface):
name="SQLiteCache",
component=symbol_cache.SqliteCache,
version=(1, 0, 0),
)
),
requirements.VersionRequirement(
name="multi_string_scanner",
component=scanners.MultiStringScanner,
version=(1, 0, 0),
),
]
@property
volatility3/framework/constants/_version.py
+2 -2
View File
@@ -1,7 +1,7 @@
# We use the SemVer 2.0.0 versioning scheme
VERSION_MAJOR = 2 # Number of releases of the library with a breaking change
VERSION_MINOR = 25 # Number of changes that only add to the interface
VERSION_PATCH = 0 # Number of changes that do not change the interface
VERSION_MINOR = 26 # Number of changes that only add to the interface
VERSION_PATCH = 1 # Number of changes that do not change the interface
VERSION_SUFFIX = ""
PACKAGE_VERSION = (
volatility3/framework/contexts/__init__.py
+1 -1
View File
@@ -130,7 +130,7 @@ class Context(interfaces.context.ContextInterface):
object_info=interfaces.objects.ObjectInformation(
layer_name=layer_name,
offset=offset,
native_layer_name=native_layer_name,
native_layer_name=native_layer_name or layer_name,
size=object_template.size,
),
)
volatility3/framework/exceptions.py
+2 -3
View File
@@ -11,7 +11,6 @@ size of the invalid page.
from typing import Callable, Dict, Optional, Tuple
from volatility3.framework import interfaces
from volatility3.framework.interfaces.configuration import VersionableInterface
class VolatilityException(Exception):
@@ -143,7 +142,7 @@ class VersionMismatchException(VolatilityException):
def __init__(
self,
source_component: Callable,
target_component: VersionableInterface,
target_component: interfaces.configuration.VersionableInterface,
target_version: Tuple[int, int, int],
failure_reason: str = None,
*args,
@@ -151,7 +150,7 @@ class VersionMismatchException(VolatilityException):
"""
Args:
source_component: The component that required the target component
target_component: The component that is required. Must inherit from VersionableInterface
target_component: The component that is required. Must inherit from interfaces.configuration.VersionableInterface
target_version: The version of the target component that was required, and ultimately was not satisfied
failure_reason: A detailed failure reason to enhance debugging and bug tracking
"""
volatility3/framework/interfaces/objects.py
+16 -31
View File
@@ -8,7 +8,7 @@ import collections
import collections.abc
import contextlib
import logging
from typing import Any, Dict, List, Mapping, Optional
from typing import Any, Dict, List, Mapping, NamedTuple, Optional
from volatility3.framework import constants, interfaces
@@ -52,7 +52,7 @@ class ReadOnlyMapping(collections.abc.Mapping):
return dict(self) == dict(other)
class ObjectInformation(ReadOnlyMapping):
class ObjectInformation(NamedTuple):
"""Contains common information useful/pertinent only to an individual
object (like an instance)
@@ -63,35 +63,20 @@ class ObjectInformation(ReadOnlyMapping):
in a single place. These values are based on the :class:`ReadOnlyMapping` class, to prevent their modification.
"""
def __init__(
self,
layer_name: str,
offset: int,
member_name: Optional[str] = None,
parent: Optional["ObjectInterface"] = None,
native_layer_name: Optional[str] = None,
size: Optional[int] = None,
):
"""Constructs a container for basic information about an object.
layer_name: str
offset: int
native_layer_name: str
member_name: Optional[str] = None
parent: Optional["ObjectInterface"] = None
size: Optional[int] = None
Args:
layer_name: Layer from which the data for the object will be read
offset: Offset within the layer at which the data for the object will be read
member_name: If the object was accessed as a member of a parent object, this was the name used to access it
parent: If the object was accessed as a member of a parent object, this is the parent object
native_layer_name: If this object references other objects (such as a pointer), what layer those objects live in
size: The size that the whole structure consumes in bytes
"""
super().__init__(
{
"layer_name": layer_name,
"offset": offset,
"member_name": member_name,
"parent": parent,
"native_layer_name": native_layer_name or layer_name,
"size": size,
}
)
def __getitem__(self, key):
if key in self._fields:
return getattr(self, key)
raise KeyError(f"NamedTuple does not have a key {key}")
def __contains__(self, key):
return key in self._fields
class ObjectInterface(metaclass=abc.ABCMeta):
@@ -183,7 +168,7 @@ class ObjectInterface(metaclass=abc.ABCMeta):
offset=self.vol.offset,
member_name=self.vol.member_name,
parent=self.vol.parent,
native_layer_name=self.vol.native_layer_name,
native_layer_name=self.vol.native_layer_name or self.vol.layer_name,
size=object_template.size,
)
return object_template(context=self._context, object_info=object_info)
volatility3/framework/interfaces/renderers.py
+74 -18
View File
@@ -10,7 +10,8 @@ suitable output.
"""
import datetime
from abc import abstractmethod, ABCMeta
import warnings
from abc import ABCMeta, abstractmethod
from collections import abc
from typing import (
Any,
@@ -20,11 +21,28 @@ from typing import (
List,
NamedTuple,
Optional,
TypeVar,
Type,
Tuple,
Type,
TypeVar,
Union,
)
from typing import Dict
from volatility3.framework import interfaces
class BasicType:
def __str__(self) -> str:
"""Fallback method for rendering basic types"""
return str(self)
class BaseAbsentValue:
"""Class that represents values which are not present for some reason."""
def __str__(self) -> str:
"""Fallback method for rendering basic types"""
return "-"
class Column(NamedTuple):
@@ -34,11 +52,37 @@ class Column(NamedTuple):
RenderOption = Any
T = TypeVar("T")
class TypeRendererInterface:
type = T
def __init__(
self, func: Optional[Callable] = None, options: Optional[Dict[str, Any]] = None
):
self._options = options or {}
setattr(self, "render", func)
@property
def options(self):
return self._options
def render(self, data: Union[T, BaseAbsentValue]) -> Any:
"""Renders a specific datatype"""
return ""
def __call__(self, data: Union[T, BaseAbsentValue]) -> Any:
"""Shortcut for render"""
return self.render(data)
class Renderer(metaclass=ABCMeta):
"""Class that defines the interface that all output renderers must
support."""
_type_renderers: Dict[Union[Type, str], Callable]
def __init__(self, options: Optional[List[RenderOption]] = None) -> None:
"""Accepts an options object to configure the renderers."""
# FIXME: Once the config option objects are in place, put the _type_check in place
@@ -102,11 +146,7 @@ class TreeNode(abc.Sequence, metaclass=ABCMeta):
"""
class BaseAbsentValue:
"""Class that represents values which are not present for some reason."""
class Disassembly:
class Disassembly(BasicType):
"""A class to indicate that the bytes provided should be disassembled
(based on the architecture)"""
@@ -115,6 +155,10 @@ class Disassembly:
def __init__(
self, data: bytes, offset: int = 0, architecture: str = "intel64"
) -> None:
warnings.warn(
"interfaces.renderers.Disassembly is now renderers.Disassembly",
FutureWarning,
)
self.data = data
self.architecture = None
if architecture in self.possible_architectures:
@@ -123,6 +167,10 @@ class Disassembly:
raise TypeError("Offset must be an integer type")
self.offset = offset
def __str__(self) -> str:
"""Fallback method of rendering"""
return str(self.data)
# We don't class these off a shared base, because the BaseTypes must only
# contain the types that the validator will accept (which would not include the base)
@@ -135,7 +183,7 @@ BaseTypes = Union[
Type[bytes],
Type[datetime.datetime],
Type[BaseAbsentValue],
Type[Disassembly],
Type[BasicType],
]
ColumnsType = List[Tuple[str, BaseTypes]]
VisitorSignature = Callable[[TreeNode, _Type], _Type]
@@ -154,16 +202,15 @@ class TreeGrid(metaclass=ABCMeta):
and to create cycles.
"""
base_types: ClassVar[Tuple] = (
int,
str,
float,
bytes,
datetime.datetime,
Disassembly,
)
# TODO: Figure out why this isn't just BaseTypes (which includes AbsentValues'
base_types: ClassVar[Tuple] = (int, str, float, bytes, datetime.datetime, BasicType)
def __init__(self, columns: ColumnsType, generator: Generator) -> None:
def __init__(
self,
columns: ColumnsType,
generator: Generator,
context: Optional["interfaces.context.ContextInterface"] = None,
) -> None:
"""Constructs a TreeGrid object using a specific set of columns.
The TreeGrid itself is a root element, that can have children but no values.
@@ -174,6 +221,15 @@ class TreeGrid(metaclass=ABCMeta):
columns: A list of column tuples made up of (name, type).
generator: An iterable containing row for a tree grid, each row contains a indent level followed by the values for each column in order.
"""
self._context = context
@property
def context(self) -> Optional["interfaces.context.ContextInterface"]:
"""Returns the context value for the tree grid (to retrieve data items)
This is a property to ensure the renderers don't try changing the context for any reason
"""
return self._context
@staticmethod
@abstractmethod
volatility3/framework/interfaces/symbols.py
+1 -2
View File
@@ -10,7 +10,6 @@ from typing import Any, Dict, Iterable, List, Mapping, Optional, Tuple, Type
from volatility3.framework import constants, exceptions, interfaces
from volatility3.framework.configuration import requirements
from volatility3.framework.interfaces import configuration, objects
from volatility3.framework.interfaces.configuration import RequirementInterface
class SymbolInterface:
@@ -347,7 +346,7 @@ class SymbolTableInterface(
return config
@classmethod
def get_requirements(cls) -> List[RequirementInterface]:
def get_requirements(cls) -> List[configuration.RequirementInterface]:
return super().get_requirements() + [
requirements.IntRequirement(
name="symbol_mask",
volatility3/framework/layers/elf.py
+2 -2
View File
@@ -6,7 +6,7 @@ import struct
from typing import Optional
from volatility3.framework import exceptions, interfaces, constants
from volatility3.framework.constants.linux import ELF_CLASS
from volatility3.framework.constants import linux as linux_constants
from volatility3.framework.layers import segmented
from volatility3.framework.symbols import intermed
@@ -23,7 +23,7 @@ class Elf64Layer(segmented.SegmentedLayer):
_header_struct = struct.Struct("<IBBB")
MAGIC = 0x464C457F # "\x7fELF"
ELF_CLASS = ELF_CLASS.ELFCLASS64
ELF_CLASS = linux_constants.ELF_CLASS.ELFCLASS64
def __init__(
self, context: interfaces.context.ContextInterface, config_path: str, name: str
volatility3/framework/layers/qemu.py
+11
View File
@@ -9,6 +9,7 @@ import struct
from typing import Any, Dict, List, Optional, Set, Tuple
from volatility3.framework import constants, exceptions, interfaces
from volatility3.framework.configuration import requirements
from volatility3.framework.layers import scanners, segmented
from volatility3.framework.symbols import intermed
@@ -99,6 +100,16 @@ class QemuSuspendLayer(segmented.NonLinearlySegmentedLayer):
context=context, config_path=config_path, name=name, metadata=metadata
)
@classmethod
def get_requirements(cls) -> List[interfaces.configuration.RequirementInterface]:
return [
requirements.VersionRequirement(
name="regex_scanner",
component=scanners.RegExScanner,
version=(1, 0, 0),
),
]
@classmethod
def _check_header(
cls, base_layer: interfaces.layers.DataLayerInterface, name: str = ""
volatility3/framework/layers/registry.py
+3 -8
View File
@@ -7,11 +7,6 @@ from typing import Any, Callable, Dict, Iterable, List, Optional, Tuple, Union
from volatility3.framework import constants, exceptions, interfaces, objects
from volatility3.framework.configuration import requirements
from volatility3.framework.configuration.requirements import (
IntRequirement,
TranslationLayerRequirement,
)
from volatility3.framework.exceptions import InvalidAddressException
from volatility3.framework.layers import linear
from volatility3.framework.symbols import intermed
from volatility3.framework.symbols.windows import extensions
@@ -154,7 +149,7 @@ class RegistryHive(linear.LinearlyMappedLayer):
@property
def root_cell_offset(self) -> int:
"""Returns the offset for the root cell in this hive."""
with contextlib.suppress(InvalidAddressException):
with contextlib.suppress(exceptions.InvalidAddressException):
if (
self._base_block.Signature.cast(
"string", max_length=4, encoding="latin-1"
@@ -271,7 +266,7 @@ class RegistryHive(linear.LinearlyMappedLayer):
@classmethod
def get_requirements(cls) -> List[interfaces.configuration.RequirementInterface]:
return [
IntRequirement(
requirements.IntRequirement(
name="hive_offset",
description="Offset within the base layer at which the hive lives",
default=0,
@@ -280,7 +275,7 @@ class RegistryHive(linear.LinearlyMappedLayer):
requirements.SymbolTableRequirement(
name="nt_symbols", description="Windows kernel symbols"
),
TranslationLayerRequirement(
requirements.TranslationLayerRequirement(
name="base_layer",
description="Layer in which the registry hive lives",
optional=False,
volatility3/framework/layers/scanners/__init__.py
+5
View File
@@ -11,6 +11,8 @@ from volatility3.framework.layers.scanners import multiregexp as multiregexp
class BytesScanner(layers.ScannerInterface):
thread_safe = True
_version = (1, 0, 0)
_required_framework_version = (2, 0, 0)
def __init__(self, needle: bytes) -> None:
@@ -38,6 +40,8 @@ class RegExScanner(layers.ScannerInterface):
thread_safe = True
_version = (1, 0, 0)
_required_framework_version = (2, 0, 0)
def __init__(self, pattern: bytes, flags: int = re.DOTALL) -> None:
@@ -57,6 +61,7 @@ class RegExScanner(layers.ScannerInterface):
class MultiStringScanner(layers.ScannerInterface):
thread_safe = True
_version = (1, 0, 0)
_required_framework_version = (2, 0, 0)
def __init__(self, patterns: List[bytes]) -> None:
volatility3/framework/layers/xen.py
+2 -2
View File
@@ -5,7 +5,7 @@ from typing import Optional
from volatility3.framework import constants, interfaces, exceptions
from volatility3.framework.layers import elf
from volatility3.framework.symbols import intermed
from volatility3.framework.constants.linux import ELF_CLASS
from volatility3.framework.constants import linux as linux_constants
vollog = logging.getLogger(__name__)
@@ -15,7 +15,7 @@ class XenCoreDumpLayer(elf.Elf64Layer):
_header_struct = struct.Struct("<IBBB")
MAGIC = 0x464C457F # "\x7fELF"
ELF_CLASS = ELF_CLASS.ELFCLASS64
ELF_CLASS = linux_constants.ELF_CLASS.ELFCLASS64
def __init__(
self, context: interfaces.context.ContextInterface, config_path: str, name: str
volatility3/framework/objects/__init__.py
+3 -2
View File
@@ -458,6 +458,7 @@ class Pointer(Integer):
offset=offset,
parent=self,
size=self.vol.subtype.size,
native_layer_name=layer_name,
),
)
return self._cache[layer_name]
@@ -811,7 +812,7 @@ class Array(interfaces.objects.ObjectInterface, collections.abc.Sequence):
layer_name=self.vol.layer_name,
offset=mask & (self.vol.offset + (self.vol.subtype.size * index)),
parent=self,
native_layer_name=self.vol.native_layer_name,
native_layer_name=self.vol.native_layer_name or self.vol.layer_name,
size=self.vol.subtype.size,
)
result += [self.vol.subtype(context=self._context, object_info=object_info)]
@@ -978,7 +979,7 @@ class AggregateType(interfaces.objects.ObjectInterface):
offset=mask & (self.vol.offset + relative_offset),
member_name=attr,
parent=self,
native_layer_name=self.vol.native_layer_name,
native_layer_name=self.vol.native_layer_name or self.vol.layer_name,
size=template.size,
)
member = template(context=self._context, object_info=object_info)
volatility3/framework/plugins/banners.py
+6 -1
View File
@@ -22,7 +22,12 @@ class Banners(interfaces.plugins.PluginInterface):
return [
requirements.TranslationLayerRequirement(
name="primary", description="Memory layer to scan"
)
),
requirements.VersionRequirement(
name="regex_scanner",
component=scanners.RegExScanner,
version=(1, 0, 0),
),
]
def _generator(self):
volatility3/framework/plugins/linux/bash.py
+12 -2
View File
@@ -13,7 +13,7 @@ from volatility3.framework.configuration import requirements
from volatility3.framework.interfaces import plugins
from volatility3.framework.layers import scanners
from volatility3.framework.objects import utility
from volatility3.framework.symbols.linux.bash import BashIntermedSymbols
from volatility3.framework.symbols.linux import bash
from volatility3.plugins import timeliner
from volatility3.plugins.linux import pslist
@@ -40,6 +40,16 @@ class Bash(plugins.PluginInterface, timeliner.TimeLinerInterface):
component=timeliner.TimeLinerInterface,
version=(1, 0, 0),
),
requirements.VersionRequirement(
name="multi_string_scanner",
component=scanners.MultiStringScanner,
version=(1, 0, 0),
),
requirements.VersionRequirement(
name="bytes_scanner",
component=scanners.BytesScanner,
version=(1, 0, 0),
),
requirements.ListRequirement(
name="pid",
element_type=int,
@@ -60,7 +70,7 @@ class Bash(plugins.PluginInterface, timeliner.TimeLinerInterface):
pack_format = "Q"
bash_json_file = "bash64"
bash_table_name = BashIntermedSymbols.create(
bash_table_name = bash.BashIntermedSymbols.create(
self.context, self.config_path, "linux", bash_json_file
)
volatility3/framework/plugins/linux/check_afinfo.py
+132 -58
View File
@@ -4,7 +4,7 @@
"""A module containing a plugin that verifies the operation function
pointers of network protocols."""
import logging
from typing import List
from typing import List, Tuple, Generator
from volatility3.framework import exceptions, interfaces
from volatility3.framework import renderers
@@ -18,6 +18,7 @@ vollog = logging.getLogger(__name__)
class Check_afinfo(plugins.PluginInterface):
"""Verifies the operation function pointers of network protocols."""
_version = (1, 0, 0)
_required_framework_version = (2, 0, 0)
@classmethod
@@ -30,61 +31,80 @@ class Check_afinfo(plugins.PluginInterface):
),
]
# returns whether the symbol is found within the kernel (system.map) or not
def _is_known_address(self, handler_addr):
symbols = list(self.context.symbol_space.get_symbols_by_location(handler_addr))
@classmethod
def _check_members(
cls,
context: interfaces.context.ContextInterface,
vmlinux_name: str,
var_ops: interfaces.objects.ObjectInterface,
var_name: str,
members: List[str],
) -> Generator[Tuple[str, str, int], None, None]:
"""
Yields any members that are not pointing inside the kernel
"""
return len(symbols) > 0
vmlinux = context.modules[vmlinux_name]
def _check_members(self, var_ops, var_name, members):
for check in members:
# redhat-specific garbage
if check.startswith("__UNIQUE_ID_rh_kabi_hide"):
continue
if check == "write":
addr = var_ops.member(attr="write")
else:
addr = getattr(var_ops, check)
# These structures have members like `write` and `next`, which are built in Python functions
addr = var_ops.member(attr=check)
if addr and addr != 0 and not self._is_known_address(addr):
yield check, addr
# Unimplemented handlers are set to 0
if not addr:
continue
def _check_afinfo(self, var_name, var, op_members, seq_members):
# check if object has a least one of the members used for analysis by this function
required_members = ["seq_fops", "seq_ops", "seq_show"]
has_required_member = any(var.has_member(member) for member in required_members)
if not has_required_member:
vollog.debug(
f"{var_name} object at {hex(var.vol.offset)} had none of the required members: {', '.join([member for member in required_members])}"
)
raise exceptions.PluginRequirementException
if len(vmlinux.get_symbols_by_absolute_location(addr)) == 0:
yield var_name, check, addr
@classmethod
def _check_pre_4_18_ops(
cls,
context: interfaces.context.ContextInterface,
vmlinux_name: str,
var_name: str,
var: interfaces.objects.ObjectInterface,
op_members: List[str],
seq_members: List[str],
):
"""
Finds the correct way to reference `op_members`
"""
vmlinux = context.modules[vmlinux_name]
if var.has_member("seq_fops"):
for hooked_member, hook_address in self._check_members(
var.seq_fops, var_name, op_members
):
yield var_name, hooked_member, hook_address
yield from cls._check_members(
context, vmlinux_name, var.seq_fops, var_name, op_members
)
# newer kernels
if var.has_member("seq_ops"):
for hooked_member, hook_address in self._check_members(
var.seq_ops, var_name, seq_members
):
yield var_name, hooked_member, hook_address
yield from cls._check_members(
context, vmlinux_name, var.seq_ops, var_name, seq_members
)
# this is the most commonly hooked member by rootkits, so a force a check on it
elif var.has_member("seq_show"):
if len(vmlinux.get_symbols_by_location(var.seq_show)) == 0:
yield var_name, "show", var.seq_show
else:
if var.has_member("seq_show"):
if not self._is_known_address(var.seq_show):
yield var_name, "show", var.seq_show
def _generator(self):
vmlinux = self.context.modules[self.config["kernel"]]
op_members = vmlinux.get_type("file_operations").members
seq_members = vmlinux.get_type("seq_operations").members
raise exceptions.VolatilityException(
"_check_afinfo_pre_4_18: Unable to find sequence operations members for checking."
)
@classmethod
def _check_afinfo_pre_4_18(
cls,
context: interfaces.context.ContextInterface,
vmlinux_name: str,
seq_members: str,
) -> Generator[Tuple[str, str, int], None, None]:
"""
Checks the operations structures for network protocols of < 4.18 systems
"""
tcp = ("tcp_seq_afinfo", ["tcp6_seq_afinfo", "tcp4_seq_afinfo"])
udp = (
"udp_seq_afinfo",
@@ -97,39 +117,93 @@ class Check_afinfo(plugins.PluginInterface):
)
protocols = [tcp, udp]
# used to track the calls to _check_afinfo and the
# number of errors produced due to missing members
symbols_checked = set()
symbols_with_errors = set()
vmlinux = context.modules[vmlinux_name]
op_members = vmlinux.get_type("file_operations").members
# loop through all symbols
for struct_type, global_vars in protocols:
for global_var_name in global_vars:
# this will lookup fail for the IPv6 protocols on kernels without IPv6 support
try:
global_var = vmlinux.get_symbol(global_var_name)
global_var = vmlinux.object_from_symbol(global_var_name)
except exceptions.SymbolError:
continue
global_var = vmlinux.object(
object_type=struct_type, offset=global_var.address
yield from cls._check_pre_4_18_ops(
context,
vmlinux_name,
global_var_name,
global_var,
op_members,
seq_members,
)
symbols_checked.add(global_var_name)
try:
for name, member, address in self._check_afinfo(
global_var_name, global_var, op_members, seq_members
):
yield 0, (name, member, format_hints.Hex(address))
except exceptions.PluginRequirementException:
symbols_with_errors.add(global_var_name)
@classmethod
def _check_afinfo_post_4_18(
cls,
context: interfaces.context.ContextInterface,
vmlinux_name: str,
seq_members: str,
) -> Generator[Tuple[str, str, int], None, None]:
"""
Checks the operations structures for network protocols of >= 4.18 systems
"""
vmlinux = context.modules[vmlinux_name]
# if every call to _check_afinfo failed show a warning
if symbols_checked == symbols_with_errors:
vollog.warning(
"This plugin was not able to check for hooks. This means you are either analyzing an unsupported kernel version or that your symbol table is corrupt."
ops_structs = [
"raw_seq_ops",
"udp_seq_ops",
"arp_seq_ops",
"unix_seq_ops",
"udp6_seq_ops",
"raw6_seq_ops",
"tcp_seq_ops",
"tcp4_seq_ops",
"tcp6_seq_ops",
"packet_seq_ops",
]
for protocol_ops_var in ops_structs:
# These will fail if the particular kernel doesn't have support for a protocol like IPv6
try:
protocol_ops = vmlinux.object_from_symbol(protocol_ops_var)
except exceptions.SymbolError:
continue
yield from cls._check_members(
context, vmlinux_name, protocol_ops, protocol_ops_var, seq_members
)
@classmethod
def check_afinfo(
cls, context: interfaces.context.ContextInterface, vmlinux_name
) -> Generator[Tuple[str, str, int], None, None]:
"""
Walks the network protocol operations structures for common network protocols.
Reports any initialized operations members that do not point inside the kernel.
"""
vmlinux = context.modules[vmlinux_name]
type_check = vmlinux.get_type("tcp_seq_afinfo")
if type_check.has_member("seq_fops"):
checker = cls._check_afinfo_pre_4_18
else:
checker = cls._check_afinfo_post_4_18
seq_members = vmlinux.get_type("seq_operations").members
yield from checker(context, vmlinux_name, seq_members)
def _generator(self):
"""
A simple wrapper around `check_afino`
"""
for name, member, address in self.check_afinfo(
self.context, self.config["kernel"]
):
yield 0, (name, member, format_hints.Hex(address))
def run(self):
return renderers.TreeGrid(
[
volatility3/framework/plugins/linux/check_modules.py
+5
View File
@@ -46,6 +46,11 @@ class Check_modules(plugins.PluginInterface):
@classmethod
def get_requirements(cls) -> List[interfaces.configuration.RequirementInterface]:
return [
requirements.VersionRequirement(
name="modules",
component=linux_utilities_modules.Modules,
version=(3, 0, 1),
),
requirements.VersionRequirement(
name="linux_utilities_modules_module_display_plugin",
component=linux_utilities_modules.ModuleDisplayPlugin,
volatility3/framework/plugins/linux/elfs.py
+2 -2
View File
@@ -14,7 +14,7 @@ from volatility3.framework.objects import utility
from volatility3.framework.renderers import format_hints
from volatility3.framework.symbols import intermed
from volatility3.framework.symbols.linux.extensions import elf
from volatility3.framework.constants.linux import ELF_MAX_EXTRACTION_SIZE
from volatility3.framework.constants import linux as linux_constants
from volatility3.plugins.linux import pslist
@@ -116,7 +116,7 @@ class Elfs(plugins.PluginInterface):
real_size = end - start
# Check if ELF has a legitimate size
if real_size < 0 or real_size > ELF_MAX_EXTRACTION_SIZE:
if real_size < 0 or real_size > linux_constants.ELF_MAX_EXTRACTION_SIZE:
raise ValueError(f"The claimed size of the ELF is invalid: {real_size}")
sections[start] = real_size
volatility3/framework/plugins/linux/hidden_modules.py
+5
View File
@@ -93,6 +93,11 @@ class Hidden_modules(plugins.PluginInterface):
component=linux_utilities_modules.ModuleDisplayPlugin,
version=(1, 0, 0),
),
requirements.VersionRequirement(
name="linux_utilities_modules",
component=linux_utilities_modules.Modules,
version=(3, 0, 1),
),
] + linux_utilities_modules.ModuleDisplayPlugin.get_requirements()
@staticmethod
volatility3/framework/plugins/linux/malfind.py
+3 -5
View File
@@ -18,7 +18,7 @@ class Malfind(interfaces.plugins.PluginInterface):
"""Lists process memory ranges that potentially contain injected code."""
_required_framework_version = (2, 0, 0)
_version = (1, 0, 2)
_version = (1, 0, 3)
@classmethod
def get_requirements(cls) -> List[interfaces.configuration.RequirementInterface]:
@@ -76,9 +76,7 @@ class Malfind(interfaces.plugins.PluginInterface):
else:
architecture = "intel64"
disasm = interfaces.renderers.Disassembly(
data, vma.vm_start, architecture
)
disasm = renderers.Disassembly(data, vma.vm_start, architecture)
yield (
0,
@@ -106,7 +104,7 @@ class Malfind(interfaces.plugins.PluginInterface):
("Path", str),
("Protection", str),
("Hexdump", format_hints.HexBytes),
("Disasm", interfaces.renderers.Disassembly),
("Disasm", renderers.Disassembly),
],
self._generator(
pslist.PsList.list_tasks(
volatility3/framework/plugins/linux/module_extract.py
+5
View File
@@ -36,6 +36,11 @@ class ModuleExtract(interfaces.plugins.PluginInterface):
description="Base virtual address to reconstruct an ELF file",
optional=False,
),
requirements.VersionRequirement(
name="linux_utilities_module_extract",
version=(1, 0, 0),
component=linux_utilities_module_extract.ModuleExtract,
),
]
def _generator(self):
volatility3/framework/plugins/linux/modxview.py
+5 -5
View File
@@ -6,9 +6,9 @@ from typing import List, Dict, Iterator
import volatility3.framework.symbols.linux.utilities.modules as linux_utilities_modules
from volatility3.framework import interfaces, deprecation
from volatility3.framework import interfaces, deprecation, renderers
from volatility3.framework.configuration import requirements
from volatility3.framework.renderers import format_hints, TreeGrid, NotAvailableValue
from volatility3.framework.renderers import format_hints
from volatility3.framework.symbols.linux import extensions
from volatility3.framework.constants import architectures
from volatility3.framework.symbols.linux.utilities import tainting
@@ -156,12 +156,12 @@ spot modules presence and taints."""
yield (
0,
(
module.get_name() or NotAvailableValue(),
module.get_name() or renderers.NotAvailableValue(),
format_hints.Hex(module_offset),
linux_utilities_modules.ModuleGathererLsmod.name in gatherers,
linux_utilities_modules.ModuleGathererSysFs.name in gatherers,
linux_utilities_modules.ModuleGathererScanner.name in gatherers,
taints or NotAvailableValue(),
taints or renderers.NotAvailableValue(),
),
)
@@ -175,7 +175,7 @@ spot modules presence and taints."""
("Taints", str),
]
return TreeGrid(
return renderers.TreeGrid(
columns,
self._generator(),
)
volatility3/framework/plugins/linux/psscan.py
+5
View File
@@ -41,6 +41,11 @@ class PsScan(interfaces.plugins.PluginInterface):
requirements.VersionRequirement(
name="pslist", component=pslist.PsList, version=(4, 0, 0)
),
requirements.VersionRequirement(
name="multi_string_scanner",
component=scanners.MultiStringScanner,
version=(1, 0, 0),
),
]
def _generator(self):
volatility3/framework/plugins/linux/sockstat.py
+12 -11
View File
@@ -5,8 +5,8 @@
import logging
from typing import Callable, Tuple, List, Dict
from volatility3.framework import interfaces, exceptions, constants, objects
from volatility3.framework.renderers import TreeGrid, NotAvailableValue, format_hints
from volatility3.framework import interfaces, exceptions, constants, objects, renderers
from volatility3.framework.renderers import format_hints
from volatility3.framework.configuration import requirements
from volatility3.framework.interfaces import plugins
from volatility3.framework.objects import utility
@@ -44,7 +44,7 @@ class SockHandlers(interfaces.configuration.VersionableInterface):
try:
netns_id = task.nsproxy.net_ns.get_inode()
except AttributeError:
netns_id = NotAvailableValue()
netns_id = renderers.NotAvailableValue()
self._netdevices = self._build_network_devices_map(netns_id)
@@ -79,7 +79,7 @@ class SockHandlers(interfaces.configuration.VersionableInterface):
)
for net_dev in net.dev_base_head.to_list(net_device_symname, "dev_list"):
if (
isinstance(netns_id, NotAvailableValue)
isinstance(netns_id, renderers.NotAvailableValue)
or net.get_inode() != netns_id
):
continue
@@ -263,7 +263,7 @@ class SockHandlers(interfaces.configuration.VersionableInterface):
# Kernel >= 3.7.10
src_port = netlink_sock.get_portid()
except AttributeError:
src_port = NotAvailableValue()
src_port = renderers.NotAvailableValue()
dst_addr = f"group:0x{netlink_sock.dst_group:08x}"
module = netlink_sock.module
@@ -273,7 +273,7 @@ class SockHandlers(interfaces.configuration.VersionableInterface):
try:
dst_port = netlink_sock.get_dst_portid()
except AttributeError:
dst_port = NotAvailableValue()
dst_port = renderers.NotAvailableValue()
state = netlink_sock.get_state()
@@ -571,7 +571,7 @@ class Sockstat(plugins.PluginInterface):
try:
netns_id = net.get_inode()
except AttributeError:
netns_id = NotAvailableValue()
netns_id = renderers.NotAvailableValue()
yield task, netns_id, fd_num, family, sock_type, protocol, sock_fields
@@ -586,10 +586,11 @@ class Sockstat(plugins.PluginInterface):
`sock_stat` and `protocol` formatted.
"""
sock_stat = [
NotAvailableValue() if field is None else str(field) for field in sock_stat
renderers.NotAvailableValue() if field is None else str(field)
for field in sock_stat
]
if protocol is None:
protocol = NotAvailableValue()
protocol = renderers.NotAvailableValue()
return tuple(sock_stat), protocol
@@ -641,7 +642,7 @@ class Sockstat(plugins.PluginInterface):
socket_filter_str = (
",".join(f"{k}={v}" for k, v in extended.items())
if extended
else NotAvailableValue()
else renderers.NotAvailableValue()
)
task_comm = utility.array_to_string(task.comm)
@@ -685,6 +686,6 @@ class Sockstat(plugins.PluginInterface):
("Filter", str),
]
return TreeGrid(
return renderers.TreeGrid(
tree_grid_args, self._generator(pids, netns_id, kernel_module_name)
)
volatility3/framework/plugins/linux/tracing/ftrace.py
+7 -7
View File
@@ -10,9 +10,9 @@ from enum import Enum
from dataclasses import dataclass
import volatility3.framework.symbols.linux.utilities.modules as linux_utilities_modules
from volatility3.framework import constants, exceptions, interfaces
from volatility3.framework import constants, exceptions, interfaces, renderers
from volatility3.framework.configuration import requirements
from volatility3.framework.renderers import format_hints, TreeGrid, NotAvailableValue
from volatility3.framework.renderers import format_hints
from volatility3.framework.constants import architectures
vollog = logging.getLogger(__name__)
@@ -239,14 +239,14 @@ class CheckFtrace(interfaces.plugins.PluginInterface):
):
formatted_results = (
format_hints.Hex(ftrace_ops_parsed.ftrace_ops_offset),
ftrace_ops_parsed.callback_symbol or NotAvailableValue(),
ftrace_ops_parsed.callback_symbol or renderers.NotAvailableValue(),
format_hints.Hex(ftrace_ops_parsed.callback_address),
ftrace_ops_parsed.hooked_symbols or NotAvailableValue(),
ftrace_ops_parsed.module_name or NotAvailableValue(),
ftrace_ops_parsed.hooked_symbols or renderers.NotAvailableValue(),
ftrace_ops_parsed.module_name or renderers.NotAvailableValue(),
(
format_hints.Hex(ftrace_ops_parsed.module_address)
if ftrace_ops_parsed.module_address is not None
else NotAvailableValue()
else renderers.NotAvailableValue()
),
)
if self.config["show_ftrace_flags"]:
@@ -266,7 +266,7 @@ class CheckFtrace(interfaces.plugins.PluginInterface):
if self.config.get("show_ftrace_flags"):
columns.append(("Flags", str))
return TreeGrid(
return renderers.TreeGrid(
columns,
self._generator(),
)
volatility3/framework/plugins/linux/tracing/tracepoints.py
+9 -9
View File
@@ -5,15 +5,15 @@
# Public researches: https://i.blackhat.com/USA21/Wednesday-Handouts/us-21-Fixing-A-Memory-Forensics-Blind-Spot-Linux-Kernel-Tracing-wp.pdf
import logging
from typing import Iterable, List, Optional
from dataclasses import dataclass
from typing import Iterable, List, Optional
import volatility3.framework.symbols.linux.utilities.modules as linux_utilities_modules
from volatility3.framework import constants, exceptions, interfaces
from volatility3.framework import constants, exceptions, interfaces, renderers
from volatility3.framework.configuration import requirements
from volatility3.framework.renderers import format_hints, NotAvailableValue, TreeGrid
from volatility3.framework.objects import utility
from volatility3.framework.constants import architectures
from volatility3.framework.objects import utility
from volatility3.framework.renderers import format_hints
vollog = logging.getLogger(__name__)
@@ -250,14 +250,14 @@ class CheckTracepoints(interfaces.plugins.PluginInterface):
formatted_results = (
tracepoint_parsed.tracepoint_name,
format_hints.Hex(tracepoint_parsed.tracepoint_address),
tracepoint_parsed.probe_name or NotAvailableValue(),
tracepoint_parsed.probe_name or renderers.NotAvailableValue(),
format_hints.Hex(tracepoint_parsed.probe_address),
tracepoint_parsed.probe_priority or NotAvailableValue(),
tracepoint_parsed.module_name or NotAvailableValue(),
tracepoint_parsed.probe_priority or renderers.NotAvailableValue(),
tracepoint_parsed.module_name or renderers.NotAvailableValue(),
(
format_hints.Hex(tracepoint_parsed.module_address)
if tracepoint_parsed.module_address is not None
else NotAvailableValue()
else renderers.NotAvailableValue()
),
)
yield (
@@ -276,7 +276,7 @@ class CheckTracepoints(interfaces.plugins.PluginInterface):
("Module address", format_hints.Hex),
]
return TreeGrid(
return renderers.TreeGrid(
columns,
self._generator(),
)
volatility3/framework/plugins/linux/vmaregexscan.py
+5
View File
@@ -46,6 +46,11 @@ class VmaRegExScan(plugins.PluginInterface):
requirements.StringRequirement(
name="pattern", description="RegEx pattern", optional=False
),
requirements.VersionRequirement(
name="regex_scanner",
component=scanners.RegExScanner,
version=(1, 0, 0),
),
requirements.IntRequirement(
name="maxsize",
description="Maximum size in bytes for displayed context",
volatility3/framework/plugins/linux/vmayarascan.py
+10 -4
View File
@@ -17,8 +17,8 @@ vollog = logging.getLogger(__name__)
class VmaYaraScan(interfaces.plugins.PluginInterface):
"""Scans all virtual memory areas for tasks using yara."""
_required_framework_version = (2, 4, 0)
_version = (1, 0, 3)
_required_framework_version = (2, 22, 0)
_version = (1, 0, 4)
@classmethod
def get_requirements(cls) -> List[interfaces.configuration.RequirementInterface]:
@@ -97,12 +97,18 @@ class VmaYaraScan(interfaces.plugins.PluginInterface):
for offset, rule_name, name, value in scanner(
proc_layer.read(start, size, pad=True), start
):
layer_data = renderers.LayerData(
context=self.context,
offset=offset,
layer_name=proc_layer.name,
length=len(value),
)
yield 0, (
format_hints.Hex(offset),
task.tgid,
rule_name,
name,
value,
layer_data,
)
@classmethod
@@ -130,7 +136,7 @@ class VmaYaraScan(interfaces.plugins.PluginInterface):
("PID", int),
("Rule", str),
("Component", str),
("Value", bytes),
("Value", renderers.LayerData),
],
self._generator(),
)
volatility3/framework/plugins/mac/bash.py
+12 -2
View File
@@ -12,7 +12,7 @@ from volatility3.framework.configuration import requirements
from volatility3.framework.interfaces import plugins
from volatility3.framework.layers import scanners
from volatility3.framework.objects import utility
from volatility3.framework.symbols.linux.bash import BashIntermedSymbols
from volatility3.framework.symbols.linux import bash
from volatility3.plugins import timeliner
from volatility3.plugins.mac import pslist
@@ -38,6 +38,16 @@ class Bash(plugins.PluginInterface, timeliner.TimeLinerInterface):
component=timeliner.TimeLinerInterface,
version=(1, 0, 0),
),
requirements.VersionRequirement(
name="multi_string_scanner",
component=scanners.MultiStringScanner,
version=(1, 0, 0),
),
requirements.VersionRequirement(
name="bytes_scanner",
component=scanners.BytesScanner,
version=(1, 0, 0),
),
requirements.ListRequirement(
name="pid",
description="Filter on specific process IDs",
@@ -58,7 +68,7 @@ class Bash(plugins.PluginInterface, timeliner.TimeLinerInterface):
pack_format = "Q"
bash_json_file = "bash64"
bash_table_name = BashIntermedSymbols.create(
bash_table_name = bash.BashIntermedSymbols.create(
self.context, self.config_path, "linux", bash_json_file
)
volatility3/framework/plugins/mac/list_files.py
+5
View File
@@ -31,6 +31,11 @@ class List_Files(plugins.PluginInterface):
requirements.VersionRequirement(
name="mount", component=mount.Mount, version=(2, 0, 0)
),
requirements.VersionRequirement(
name="mac_utilities",
component=mac.MacUtilities,
version=(1, 3, 0),
),
]
@classmethod
volatility3/framework/plugins/mac/malfind.py
+3 -5
View File
@@ -13,7 +13,7 @@ from volatility3.plugins.mac import pslist
class Malfind(interfaces.plugins.PluginInterface):
"""Lists process memory ranges that potentially contain injected code."""
_required_framework_version = (2, 0, 0)
_required_framework_version = (2, 0, 1)
@classmethod
def get_requirements(cls):
@@ -68,9 +68,7 @@ class Malfind(interfaces.plugins.PluginInterface):
else:
architecture = "intel64"
disasm = interfaces.renderers.Disassembly(
data, vma.links.start, architecture
)
disasm = renderers.Disassembly(data, vma.links.start, architecture)
yield (
0,
@@ -99,7 +97,7 @@ class Malfind(interfaces.plugins.PluginInterface):
("End", format_hints.Hex),
("Protection", str),
("Hexdump", format_hints.HexBytes),
("Disasm", interfaces.renderers.Disassembly),
("Disasm", renderers.Disassembly),
],
self._generator(
list_tasks(self.context, self.config["kernel"], filter_func=filter_func)
volatility3/framework/plugins/regexscan.py
+5
View File
@@ -39,6 +39,11 @@ class RegExScan(plugins.PluginInterface):
default=cls.MAXSIZE_DEFAULT,
optional=True,
),
requirements.VersionRequirement(
name="regex_scanner",
component=scanners.RegExScanner,
version=(1, 0, 0),
),
]
def _generator(self, regex_pattern):
volatility3/framework/plugins/vmscan.py
+7
View File
@@ -26,6 +26,8 @@ class VMCSTest(enum.IntFlag):
class PageStartScanner(interfaces.layers.ScannerInterface):
_version = (1, 0, 0)
def __init__(self, signatures: List[bytes], page_size: int = 0x1000):
super().__init__()
if not len(signatures):
@@ -69,6 +71,11 @@ class Vmscan(plugins.PluginInterface):
requirements.TranslationLayerRequirement(
name="primary", description="Physical base memory layer"
),
requirements.VersionRequirement(
name="page_start_scanner",
component=PageStartScanner,
version=(1, 0, 0),
),
requirements.IntRequirement(
name="log-threshold",
description="Number of criteria failed to log to debug output",
volatility3/framework/plugins/windows/cmdscan.py
+5
View File
@@ -41,6 +41,11 @@ class CmdScan(interfaces.plugins.PluginInterface):
requirements.VersionRequirement(
name="consoles", component=consoles.Consoles, version=(3, 0, 0)
),
requirements.VersionRequirement(
name="bytes_scanner",
component=scanners.BytesScanner,
version=(1, 0, 0),
),
requirements.BooleanRequirement(
name="no_registry",
description="Don't search the registry for possible values of CommandHistorySize",
volatility3/framework/plugins/windows/consoles.py
+5
View File
@@ -54,6 +54,11 @@ class Consoles(interfaces.plugins.PluginInterface):
requirements.VersionRequirement(
name="hivelist", component=hivelist.HiveList, version=(2, 0, 0)
),
requirements.VersionRequirement(
name="bytes_scanner",
component=scanners.BytesScanner,
version=(1, 0, 0),
),
requirements.BooleanRequirement(
name="no_registry",
description="Don't search the registry for possible values of CommandHistorySize and HistoryBufferMax",
volatility3/framework/plugins/windows/dumpfiles.py
+9 -4
View File
@@ -7,9 +7,14 @@ import ntpath
import re
from typing import List, Tuple, Type, Optional, Generator
from volatility3.framework import interfaces, renderers, exceptions, constants
from volatility3.framework import (
interfaces,
exceptions,
constants,
renderers,
)
from volatility3.framework.configuration import requirements
from volatility3.framework.renderers import format_hints, UnreadableValue
from volatility3.framework.renderers import format_hints
from volatility3.plugins.windows import handles
from volatility3.plugins.windows import pslist
@@ -258,7 +263,7 @@ class DumpFiles(interfaces.plugins.PluginInterface):
if file_re:
name = file_obj.file_name_with_device()
if isinstance(name, UnreadableValue):
if isinstance(name, renderers.UnreadableValue):
continue
if not file_re.search(name):
continue
@@ -298,7 +303,7 @@ class DumpFiles(interfaces.plugins.PluginInterface):
if file_re:
name = file_obj.file_name_with_device()
if isinstance(name, UnreadableValue):
if isinstance(name, renderers.UnreadableValue):
continue
if not file_re.search(name):
continue
volatility3/framework/plugins/windows/info.py
+5 -4
View File
@@ -3,12 +3,11 @@
#
import time
from typing import List, Tuple, Iterable
from typing import Iterable, List, Tuple
from volatility3.framework import constants, interfaces, layers, symbols
from volatility3.framework import constants, interfaces, layers, renderers, symbols
from volatility3.framework.configuration import requirements
from volatility3.framework.interfaces import plugins
from volatility3.framework.renderers import TreeGrid
from volatility3.framework.symbols import intermed
from volatility3.framework.symbols.windows.extensions import kdbg, pe
@@ -294,4 +293,6 @@ class Info(plugins.PluginInterface):
)
def run(self):
return TreeGrid([("Variable", str), ("Value", str)], self._generator())
return renderers.TreeGrid(
[("Variable", str), ("Value", str)], self._generator()
)
volatility3/framework/plugins/windows/malfind.py
+41 -12
View File
@@ -2,7 +2,7 @@
# which is available at https://www.volatilityfoundation.org/license/vsl-v1.0
#
import logging
from typing import Iterable, Tuple
from typing import Iterable, Generator, Tuple
from volatility3.framework import interfaces, symbols, exceptions
from volatility3.framework import renderers
@@ -17,7 +17,8 @@ vollog = logging.getLogger(__name__)
class Malfind(interfaces.plugins.PluginInterface):
"""Lists process memory ranges that potentially contain injected code."""
_required_framework_version = (2, 4, 0)
_required_framework_version = (2, 22, 0)
_version = (1, 1, 0)
@classmethod
def get_requirements(cls):
@@ -88,6 +89,25 @@ class Malfind(interfaces.plugins.PluginInterface):
symbol_table: str,
proc: interfaces.objects.ObjectInterface,
) -> Iterable[Tuple[interfaces.objects.ObjectInterface, bytes]]:
for vad, data_object in cls.list_injection_sites(
context, kernel_layer_name, symbol_table, proc
):
yield vad, data_object.context.layers[data_object.layer_name].read(
data_object.offset, data_object.length
)
@classmethod
def list_injection_sites(
cls,
context: interfaces.context.ContextInterface,
kernel_layer_name: str,
symbol_table: str,
proc: interfaces.objects.ObjectInterface,
) -> Generator[
Tuple[interfaces.objects.ObjectInterface, renderers.LayerData],
None,
None,
]:
"""Generate memory regions for a process that may contain injected
code.
@@ -156,8 +176,16 @@ class Malfind(interfaces.plugins.PluginInterface):
vollog.warning(
f"[proc_id {proc_id}] Found suspicious DIRTY + {protection_string} page at {hex(dirty_page)}",
)
data = proc_layer.read(vad.get_start(), 64, pad=True)
yield vad, data
start = vad.get_start()
length = 64
data = renderers.LayerData(
context=context,
layer_name=proc_layer_name,
offset=start,
length=length,
no_surrounding=True,
)
yield (vad, data)
def _generator(self, procs):
# determine if we're on a 32 or 64 bit kernel
@@ -166,7 +194,7 @@ class Malfind(interfaces.plugins.PluginInterface):
# set refined criteria to know when to add to "Notes" column
refined_criteria = {
b"MZ": "MZ header",
b"\x55\x8B": "PE header",
b"\x55\x8b": "PE header",
b"\x55\x48": "Function prologue",
b"\x55\x89": "Function prologue",
}
@@ -179,11 +207,14 @@ class Malfind(interfaces.plugins.PluginInterface):
# by default, "Notes" column will be set to N/A
process_name = utility.array_to_string(proc.ImageFileName)
for vad, data in self.list_injections(
for vad, data_object in self.list_injection_sites(
self.context, kernel.layer_name, kernel.symbol_table_name, proc
):
notes = renderers.NotApplicableValue()
# Check for unique headers and update "Notes" column if criteria is met
data = data_object.context.layers[data_object.layer_name].read(
data_object.offset, data_object.length, True
)
if data[0:2] in refined_criteria:
notes = refined_criteria[data[0:2]]
@@ -193,9 +224,7 @@ class Malfind(interfaces.plugins.PluginInterface):
else:
architecture = "intel64"
disasm = interfaces.renderers.Disassembly(
data, vad.get_start(), architecture
)
disasm = renderers.Disassembly(data, vad.get_start(), architecture)
file_output = "Disabled"
if self.config["dump"]:
@@ -231,7 +260,7 @@ class Malfind(interfaces.plugins.PluginInterface):
vad.get_private_memory(),
file_output,
notes,
format_hints.HexBytes(data),
data_object,
disasm,
),
)
@@ -251,8 +280,8 @@ class Malfind(interfaces.plugins.PluginInterface):
("PrivateMemory", int),
("File output", str),
("Notes", str),
("Hexdump", format_hints.HexBytes),
("Disasm", interfaces.renderers.Disassembly),
("Hexdump", renderers.LayerData),
("Disasm", renderers.Disassembly),
],
self._generator(
pslist.PsList.list_processes(
volatility3/framework/plugins/windows/mbrscan.py
+19 -12
View File
@@ -20,8 +20,8 @@ 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)
_required_framework_version = (2, 22, 0)
_version = (1, 0, 1)
@classmethod
def get_requirements(cls) -> List[interfaces.configuration.RequirementInterface]:
@@ -37,6 +37,11 @@ class MBRScan(interfaces.plugins.PluginInterface):
default=False,
optional=True,
),
requirements.VersionRequirement(
name="multi_string_scanner",
component=scanners.MultiStringScanner,
version=(1, 0, 0),
),
]
@classmethod
@@ -74,7 +79,7 @@ class MBRScan(interfaces.plugins.PluginInterface):
partition_table_object = symbol_table + constants.BANG + "PARTITION_TABLE"
# Define Signature and Data Length
mbr_signature = b"\x55\xAA"
mbr_signature = b"\x55\xaa"
mbr_length = 0x200
bootcode_length = 0x1B8
@@ -120,9 +125,7 @@ class MBRScan(interfaces.plugins.PluginInterface):
renderers.NotApplicableValue(),
renderers.NotApplicableValue(),
renderers.NotApplicableValue(),
interfaces.renderers.Disassembly(
bootcode, 0, architecture
),
renderers.Disassembly(bootcode, 0, architecture),
),
)
else:
@@ -146,10 +149,14 @@ class MBRScan(interfaces.plugins.PluginInterface):
renderers.NotApplicableValue(),
renderers.NotApplicableValue(),
renderers.NotApplicableValue(),
interfaces.renderers.Disassembly(
bootcode, 0, architecture
renderers.Disassembly(bootcode, 0, architecture),
renderers.LayerData(
context=self.context,
layer_name=layer.name,
offset=mbr_start_offset,
length=bootcode_length,
no_surrounding=True,
),
format_hints.HexBytes(bootcode),
),
)
@@ -232,7 +239,7 @@ class MBRScan(interfaces.plugins.PluginInterface):
("Bootable", bool),
("PartitionType", str),
("SectorInSize", format_hints.Hex),
("Disasm", interfaces.renderers.Disassembly),
("Disasm", renderers.Disassembly),
],
self._generator(),
)
@@ -256,8 +263,8 @@ class MBRScan(interfaces.plugins.PluginInterface):
("EndingCHS", int),
("EndingSector", int),
("SectorInSize", format_hints.Hex),
("Disasm", interfaces.renderers.Disassembly),
("Bootcode", format_hints.HexBytes),
("Disasm", renderers.Disassembly),
("Bootcode", renderers.LayerData),
],
self._generator(),
)
volatility3/framework/plugins/windows/mftscan.py
+7 -7
View File
@@ -231,7 +231,7 @@ class MFTScan(interfaces.plugins.PluginInterface, timeliner.TimeLinerInterface):
content = attr.get_resident_filecontent()
if content:
content = format_hints.HexBytes(content)
content = renderers.LayerData.from_object(content)
else:
content = renderers.NotAvailableValue()
@@ -334,9 +334,9 @@ class MFTScan(interfaces.plugins.PluginInterface, timeliner.TimeLinerInterface):
class ADS(interfaces.plugins.PluginInterface):
"""Scans for Alternate Data Stream"""
_required_framework_version = (2, 7, 0)
_required_framework_version = (2, 22, 0)
_version = (1, 0, 1)
_version = (1, 0, 2)
@classmethod
def get_requirements(cls):
@@ -395,7 +395,7 @@ class ADS(interfaces.plugins.PluginInterface):
("MFT Type", str),
("Filename", str),
("ADS Filename", str),
("Hexdump", format_hints.HexBytes),
("Hexdump", renderers.LayerData),
],
self._generator(),
)
@@ -404,9 +404,9 @@ class ADS(interfaces.plugins.PluginInterface):
class ResidentData(interfaces.plugins.PluginInterface):
"""Scans for MFT Records with Resident Data"""
_required_framework_version = (2, 7, 0)
_required_framework_version = (2, 22, 0)
_version = (1, 0, 1)
_version = (1, 0, 2)
@classmethod
def get_requirements(cls):
@@ -461,7 +461,7 @@ class ResidentData(interfaces.plugins.PluginInterface):
("Record Number", int),
("MFT Type", str),
("Filename", str),
("Hexdump", format_hints.HexBytes),
("Hexdump", renderers.LayerData),
],
self._generator(),
)
volatility3/framework/plugins/windows/pe_symbols.py
+2 -2
View File
@@ -17,7 +17,7 @@ from volatility3.framework.symbols import intermed
from volatility3.framework.symbols.windows import pdbutil
from volatility3.framework.symbols.windows.extensions import pe
from volatility3.plugins.windows import pslist, modules
from volatility3.framework.constants.windows import KERNEL_MODULE_NAMES
from volatility3.framework.constants import windows
vollog = logging.getLogger(__name__)
@@ -533,7 +533,7 @@ class PESymbols(interfaces.plugins.PluginInterface):
# a `ntoskrnl.exe` can have an internal PDB name of any of the ones in the following list
# The code attempts to find all possible PDBs to ensure the best chance of recovery
if mod_name == PESymbols.os_module_name:
pdb_names = [fn + ".pdb" for fn in KERNEL_MODULE_NAMES]
pdb_names = [fn + ".pdb" for fn in windows.KERNEL_MODULE_NAMES]
# for non-kernel files, replace the exe, sys, or dll extension with pdb
else:
volatility3/framework/plugins/windows/poolscanner.py
+7
View File
@@ -55,6 +55,8 @@ class PoolConstraint:
class PoolHeaderScanner(interfaces.layers.ScannerInterface):
_version = (1, 0, 0)
def __init__(
self,
module: interfaces.context.ModuleInterface,
@@ -142,6 +144,11 @@ class PoolScanner(plugins.PluginInterface):
requirements.VersionRequirement(
name="handles", component=handles.Handles, version=(3, 0, 0)
),
requirements.VersionRequirement(
name="pool_header_scanner",
component=PoolHeaderScanner,
version=(1, 0, 0),
),
]
def _generator(self):
volatility3/framework/plugins/windows/psxview.py
+3 -3
View File
@@ -4,10 +4,10 @@ import string
from itertools import chain
from typing import Dict, Iterable, List
from volatility3.framework import constants, exceptions
from volatility3.framework import constants, exceptions, renderers
from volatility3.framework.configuration import requirements
from volatility3.framework.interfaces import plugins
from volatility3.framework.renderers import TreeGrid, format_hints
from volatility3.framework.renderers import format_hints
from volatility3.framework.symbols.windows import extensions
from volatility3.plugins.windows import (
handles,
@@ -231,7 +231,7 @@ We recommend using -r pretty if you are looking at this plugin's output in a ter
offset_type = "(Physical)" if self.config["physical-offsets"] else "(Virtual)"
offset_str = "Offset" + offset_type
return TreeGrid(
return renderers.TreeGrid(
[
(offset_str, format_hints.Hex),
("Name", str),
volatility3/framework/plugins/windows/registry/hashdump.py
+15 -13
View File
@@ -11,8 +11,7 @@ from Crypto.Cipher import AES, ARC4, DES
from volatility3.framework import interfaces, renderers, exceptions, constants
from volatility3.framework.configuration import requirements
from volatility3.framework.exceptions import InvalidAddressException
from volatility3.framework.layers import registry as registrylayer
from volatility3.framework.layers import registry as registry_layer
from volatility3.framework.symbols.windows.extensions import registry
from volatility3.plugins.windows.registry import hivelist
@@ -329,13 +328,13 @@ class Hashdump(interfaces.plugins.PluginInterface):
@classmethod
def get_hive_key(
cls, hive: registry.RegistryHive, key: str
cls, hive: registry_layer.RegistryHive, key: str
) -> Optional["registry.CM_KEY_NODE"]:
result = None
try:
if hive:
result = hive.get_key(key)
except (KeyError, registrylayer.RegistryException):
except (KeyError, registry_layer.RegistryException):
vollog.info(
f"Unable to load the required registry key {hive.get_name()}\\{key} from this memory image"
)
@@ -343,7 +342,7 @@ class Hashdump(interfaces.plugins.PluginInterface):
@classmethod
def get_user_keys(
cls, samhive: registry.RegistryHive
cls, samhive: registry_layer.RegistryHive
) -> List[interfaces.objects.ObjectInterface]:
user_key_path = "SAM\\Domains\\Account\\Users"
@@ -354,7 +353,7 @@ class Hashdump(interfaces.plugins.PluginInterface):
return [k for k in user_key.get_subkeys() if k.Name != "Names"]
@classmethod
def get_bootkey(cls, syshive: registry.RegistryHive) -> Optional[bytes]:
def get_bootkey(cls, syshive: registry_layer.RegistryHive) -> Optional[bytes]:
"""
Returns the scrambled bootkey necesary to decrypt hashes
"""
@@ -382,8 +381,8 @@ class Hashdump(interfaces.plugins.PluginInterface):
return None
bootkey += class_data.decode("utf-16-le")
except (
InvalidAddressException,
registrylayer.RegistryException,
exceptions.InvalidAddressException,
registry_layer.RegistryException,
) as excp:
vollog.log(
constants.LOGLEVEL_VVV, f"Unable to read Lsa key {lk}: {excp}"
@@ -398,7 +397,7 @@ class Hashdump(interfaces.plugins.PluginInterface):
@classmethod
def get_hbootkey(
cls, samhive: registry.RegistryHive, bootkey: bytes
cls, samhive: registry_layer.RegistryHive, bootkey: bytes
) -> Optional[bytes]:
sam_account_path = "SAM\\Domains\\Account"
@@ -456,7 +455,10 @@ class Hashdump(interfaces.plugins.PluginInterface):
@classmethod
def get_user_hashes(
cls, user: registry.CM_KEY_NODE, samhive: registry.RegistryHive, hbootkey: bytes
cls,
user: registry.CM_KEY_NODE,
samhive: registry_layer.RegistryHive,
hbootkey: bytes,
) -> Optional[Tuple[bytes, bytes]]:
## Will sometimes find extra user with rid = NAMES, returns empty strings right now
try:
@@ -470,7 +472,7 @@ class Hashdump(interfaces.plugins.PluginInterface):
sam_data = samhive.read(v.Data + 4, v.DataLength)
except (
exceptions.InvalidAddressException,
registrylayer.RegistryException,
registry_layer.RegistryException,
):
return None
@@ -570,7 +572,7 @@ class Hashdump(interfaces.plugins.PluginInterface):
@classmethod
def get_user_name(
cls, user: registry.CM_KEY_NODE, samhive: registry.RegistryHive
cls, user: registry.CM_KEY_NODE, samhive: registry_layer.RegistryHive
) -> Optional[bytes]:
value = None
for v in user.get_values():
@@ -593,7 +595,7 @@ class Hashdump(interfaces.plugins.PluginInterface):
# replaces the dump_hashes method in vol2
def _generator(
self, syshive: registry.RegistryHive, samhive: registry.RegistryHive
self, syshive: registry_layer.RegistryHive, samhive: registry_layer.RegistryHive
):
if syshive is None:
vollog.debug("SYSTEM address is None: No system hive found")
volatility3/framework/plugins/windows/registry/lsadump.py
+12 -11
View File
@@ -10,9 +10,8 @@ from Crypto.Cipher import ARC4, DES, AES
from volatility3.framework import interfaces, renderers, exceptions
from volatility3.framework.configuration import requirements
from volatility3.framework.exceptions import InvalidAddressException
from volatility3.framework.layers import registry
from volatility3.framework.layers import registry as registry_layer
from volatility3.framework.symbols.windows import versions
from volatility3.plugins.windows.registry import hashdump, hivelist
from volatility3.framework.renderers import format_hints
@@ -65,7 +64,7 @@ class Lsadump(interfaces.plugins.PluginInterface):
@classmethod
def get_lsa_key(
cls, sechive: registry.RegistryHive, bootkey: bytes, vista_or_later: bool
cls, sechive: registry_layer.RegistryHive, bootkey: bytes, vista_or_later: bool
) -> Optional[bytes]:
if not bootkey:
return None
@@ -109,7 +108,7 @@ class Lsadump(interfaces.plugins.PluginInterface):
@classmethod
def get_secret_by_name(
cls,
sechive: registry.RegistryHive,
sechive: registry_layer.RegistryHive,
name: str,
lsakey: bytes,
is_vista_or_later: bool,
@@ -123,8 +122,8 @@ class Lsadump(interfaces.plugins.PluginInterface):
try:
enc_secret_value = next(enc_secret_key.get_values(), None)
except (
InvalidAddressException,
registry.RegistryException,
exceptions.InvalidAddressException,
registry_layer.RegistryException,
):
enc_secret_value = None
@@ -171,7 +170,9 @@ class Lsadump(interfaces.plugins.PluginInterface):
return decrypted_data[8 : 8 + dec_data_len]
def _generator(
self, syshive: registry.RegistryHive, sechive: registry.RegistryHive
self,
syshive: registry_layer.RegistryHive,
sechive: registry_layer.RegistryHive,
):
kernel = self.context.modules[self.config["kernel"]]
@@ -206,8 +207,8 @@ class Lsadump(interfaces.plugins.PluginInterface):
enc_secret_value = next(sec_val_key.get_values(), None)
except (
StopIteration,
InvalidAddressException,
registry.RegistryException,
exceptions.InvalidAddressException,
registry_layer.RegistryException,
):
enc_secret_value = None
@@ -229,8 +230,8 @@ class Lsadump(interfaces.plugins.PluginInterface):
try:
key_name = key.get_name()
except (
InvalidAddressException,
registry.RegistryException,
exceptions.InvalidAddressException,
registry_layer.RegistryException,
):
key_name = renderers.UnreadableValue()
volatility3/framework/plugins/windows/registry/printkey.py
+27 -26
View File
@@ -4,18 +4,13 @@
import datetime
import logging
from typing import List, Optional, Sequence, Iterable, Tuple, Union
from typing import Iterable, List, Optional, Sequence, Tuple, Union
from volatility3.framework import objects, renderers, exceptions, interfaces, constants
from volatility3.framework import constants, exceptions, interfaces, objects, renderers
from volatility3.framework.configuration import requirements
from volatility3.framework.layers.registry import (
RegistryHive,
RegistryFormatException,
InvalidAddressException,
RegistryException,
)
from volatility3.framework.renderers import TreeGrid, conversion, format_hints
from volatility3.framework.symbols.windows.extensions.registry import RegValueTypes
from volatility3.framework.layers import registry as registry_layer
from volatility3.framework.renderers import conversion, format_hints
from volatility3.framework.symbols.windows.extensions import registry
from volatility3.plugins.windows.registry import hivelist
vollog = logging.getLogger(__name__)
@@ -55,7 +50,7 @@ class PrintKey(interfaces.plugins.PluginInterface):
@classmethod
def key_iterator(
cls,
hive: RegistryHive,
hive: registry_layer.RegistryHive,
node_path: Optional[Sequence[objects.StructType]] = None,
recurse: bool = False,
) -> Iterable[
@@ -87,14 +82,14 @@ class PrintKey(interfaces.plugins.PluginInterface):
try:
key_path_names.append(k.get_name())
except (
InvalidAddressException,
RegistryException,
registry_layer.InvalidAddressException,
registry_layer.RegistryException,
):
key_path_names.append("-")
key_path = "\\".join([k for k in key_path_names])
if node.vol.type_name.endswith(constants.BANG + "_CELL_DATA"):
raise RegistryFormatException(
raise registry_layer.RegistryFormatException(
hive.name, "Encountered _CELL_DATA instead of _CM_KEY_NODE"
)
last_write_time = conversion.wintime_to_datetime(node.LastWriteTime.QuadPart)
@@ -116,7 +111,7 @@ class PrintKey(interfaces.plugins.PluginInterface):
key_node.get_name()
except (
exceptions.InvalidAddressException,
RegistryException,
registry_layer.RegistryException,
) as excp:
vollog.debug(excp)
continue
@@ -138,7 +133,7 @@ class PrintKey(interfaces.plugins.PluginInterface):
def _printkey_iterator(
self,
hive: RegistryHive,
hive: registry_layer.RegistryHive,
node_path: Optional[Sequence[objects.StructType]] = None,
recurse: bool = False,
):
@@ -166,7 +161,7 @@ class PrintKey(interfaces.plugins.PluginInterface):
key_node_name = node.get_name()
except (
exceptions.InvalidAddressException,
RegistryException,
registry_layer.RegistryException,
) as excp:
vollog.debug(excp)
key_node_name = renderers.UnreadableValue()
@@ -193,16 +188,16 @@ class PrintKey(interfaces.plugins.PluginInterface):
value_node_name = node.get_name() or "(Default)"
except (
exceptions.InvalidAddressException,
RegistryException,
registry_layer.RegistryException,
) as excp:
vollog.debug(excp)
value_node_name = renderers.UnreadableValue()
try:
value_type = RegValueTypes(node.Type).name
value_type = registry.RegValueTypes(node.Type).name
except (
exceptions.InvalidAddressException,
RegistryException,
registry_layer.RegistryException,
) as excp:
vollog.debug(excp)
value_type = renderers.UnreadableValue()
@@ -222,11 +217,17 @@ class PrintKey(interfaces.plugins.PluginInterface):
value_data = format_hints.MultiTypeData(
value_data, encoding="utf-8"
)
elif RegValueTypes(node.Type) == RegValueTypes.REG_BINARY:
elif (
registry.RegValueTypes(node.Type)
== registry.RegValueTypes.REG_BINARY
):
value_data = format_hints.MultiTypeData(
value_data, show_hex=True
)
elif RegValueTypes(node.Type) == RegValueTypes.REG_MULTI_SZ:
elif (
registry.RegValueTypes(node.Type)
== registry.RegValueTypes.REG_MULTI_SZ
):
value_data = format_hints.MultiTypeData(
value_data, encoding="utf-16-le", split_nulls=True
)
@@ -237,7 +238,7 @@ class PrintKey(interfaces.plugins.PluginInterface):
except (
ValueError,
exceptions.InvalidAddressException,
RegistryException,
registry_layer.RegistryException,
) as excp:
vollog.debug(excp)
value_data = renderers.UnreadableValue()
@@ -279,13 +280,13 @@ class PrintKey(interfaces.plugins.PluginInterface):
except (
exceptions.InvalidAddressException,
KeyError,
RegistryException,
registry_layer.RegistryException,
) as excp:
if isinstance(excp, KeyError):
vollog.debug(
f"Key '{key}' not found in Hive at offset {hex(hive.hive_offset)}."
)
elif isinstance(excp, RegistryException):
elif isinstance(excp, registry_layer.RegistryException):
vollog.debug(excp)
elif isinstance(excp, exceptions.InvalidAddressException):
vollog.debug(
@@ -308,7 +309,7 @@ class PrintKey(interfaces.plugins.PluginInterface):
def run(self):
offset = self.config.get("offset", None)
return TreeGrid(
return renderers.TreeGrid(
columns=[
("Last Write Time", datetime.datetime),
("Hive Offset", format_hints.Hex),
volatility3/framework/plugins/windows/registry/scheduled_tasks.py
+1 -1
View File
@@ -1216,7 +1216,7 @@ class ScheduledTasks(interfaces.plugins.PluginInterface, timeliner.TimeLinerInte
@classmethod
def _get_task_keys(
cls, software_hive: reg_extensions.RegistryHive
cls, software_hive: registry.RegistryHive
) -> Tuple[
Optional[reg_extensions.CM_KEY_NODE], Optional[reg_extensions.CM_KEY_NODE]
]:
volatility3/framework/plugins/windows/registry/userassist.py
+7 -10
View File
@@ -12,11 +12,8 @@ from typing import Any, Generator, List, Tuple
from volatility3.framework import constants, exceptions, interfaces, renderers
from volatility3.framework.configuration import requirements
from volatility3.framework.layers.physical import BufferDataLayer
from volatility3.framework.layers.registry import (
RegistryHive,
RegistryException,
)
from volatility3.framework.layers import physical
from volatility3.framework.layers import registry as registry_layer
from volatility3.framework.renderers import conversion, format_hints
from volatility3.framework.symbols import intermed
from volatility3.plugins.windows.registry import hivelist
@@ -94,7 +91,7 @@ class UserAssist(interfaces.plugins.PluginInterface, timeliner.TimeLinerInterfac
return item
userassist_layer_name = self.context.layers.free_layer_name("userassist_buffer")
buffer = BufferDataLayer(
buffer = physical.BufferDataLayer(
self.context, self._config_path, userassist_layer_name, userassist_data
)
self.context.add_layer(buffer)
@@ -158,7 +155,7 @@ class UserAssist(interfaces.plugins.PluginInterface, timeliner.TimeLinerInterfac
).has_member("CookiePad")
def list_userassist(
self, hive: RegistryHive
self, hive: registry_layer.RegistryHive
) -> Generator[Tuple[int, Tuple], None, None]:
"""Generate userassist data for a registry hive."""
@@ -180,7 +177,7 @@ class UserAssist(interfaces.plugins.PluginInterface, timeliner.TimeLinerInterfac
"software\\microsoft\\windows\\currentversion\\explorer\\userassist",
return_list=True,
)
except RegistryException as e:
except registry_layer.RegistryException as e:
vollog.warning(
f"Error accessing UserAssist key in {hive_name} at {hive.hive_offset:#x}: {e}"
)
@@ -250,7 +247,7 @@ class UserAssist(interfaces.plugins.PluginInterface, timeliner.TimeLinerInterfac
subkey_name = subkey.get_name()
except (
exceptions.InvalidAddressException,
RegistryException,
registry_layer.RegistryException,
):
subkey_name = renderers.UnreadableValue()
@@ -279,7 +276,7 @@ class UserAssist(interfaces.plugins.PluginInterface, timeliner.TimeLinerInterfac
value_name = value.get_name()
except (
exceptions.InvalidAddressException,
RegistryException,
registry_layer.RegistryException,
):
value_name = renderers.UnreadableValue()
volatility3/framework/plugins/windows/skeleton_key_check.py
+5
View File
@@ -63,6 +63,11 @@ class Skeleton_Key_Check(interfaces.plugins.PluginInterface):
requirements.VersionRequirement(
name="pe_symbols", component=pe_symbols.PESymbols, version=(3, 0, 0)
),
requirements.VersionRequirement(
name="bytes_scanner",
component=scanners.BytesScanner,
version=(1, 0, 0),
),
]
def _check_for_skeleton_key_vad(
volatility3/framework/plugins/windows/svclist.py
+5
View File
@@ -42,6 +42,11 @@ class SvcList(svcscan.SvcScan):
description="Windows kernel",
architectures=["Intel32", "Intel64"],
),
requirements.VersionRequirement(
name="bytes_scanner",
component=scanners.BytesScanner,
version=(1, 0, 0),
),
]
@classmethod
volatility3/framework/plugins/windows/svcscan.py
+5
View File
@@ -56,6 +56,11 @@ class SvcScan(interfaces.plugins.PluginInterface):
requirements.VersionRequirement(
name="hivelist", component=hivelist.HiveList, version=(2, 0, 0)
),
requirements.VersionRequirement(
name="bytes_scanner",
component=scanners.BytesScanner,
version=(1, 0, 0),
),
]
@classmethod
volatility3/framework/plugins/windows/truecrypt.py
+12 -17
View File
@@ -3,20 +3,15 @@
#
import logging
from typing import Generator, Iterable, List, Tuple
from typing import Iterable, Generator, List, Tuple
from volatility3.framework import constants, interfaces, renderers
from volatility3.framework import constants, interfaces, objects, renderers
from volatility3.framework.configuration import requirements
from volatility3.framework.interfaces.configuration import RequirementInterface
from volatility3.framework.interfaces.objects import ObjectInterface
from volatility3.framework.objects import Bytes, DataFormatInfo, Integer, StructType
from volatility3.framework.objects.templates import ObjectTemplate
from volatility3.framework.interfaces import configuration
from volatility3.framework.objects.utility import array_to_string
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 modules
vollog = logging.getLogger(__name__)
@@ -29,7 +24,7 @@ class Passphrase(interfaces.plugins.PluginInterface):
_required_framework_version = (2, 5, 2)
@classmethod
def get_requirements(cls) -> List[RequirementInterface]:
def get_requirements(cls) -> List[configuration.RequirementInterface]:
return [
requirements.ModuleRequirement(
"kernel",
@@ -67,7 +62,7 @@ class Passphrase(interfaces.plugins.PluginInterface):
layer_name,
module_base,
)
data_section: StructType = next(
data_section: objects.StructType = next(
sec
for sec in dos_header.get_nt_header().get_sections()
if array_to_string(sec.Name) == ".data"
@@ -76,11 +71,11 @@ class Passphrase(interfaces.plugins.PluginInterface):
size: int = data_section.Misc.VirtualSize
# Looking at `Length` in TrueCrypt/Common/Password.h::Password struct
DWORD_SIZE_BYTES: int = 4
format = DataFormatInfo(
format = objects.DataFormatInfo(
length=DWORD_SIZE_BYTES, byteorder="little", signed=True
)
int32 = ObjectTemplate(
Integer, pe_table_name + constants.BANG + "int", data_format=format
int32 = objects.templates.ObjectTemplate(
objects.Integer, pe_table_name + constants.BANG + "int", data_format=format
)
count, not_aligned = divmod(size, DWORD_SIZE_BYTES)
if not_aligned:
@@ -99,7 +94,7 @@ class Passphrase(interfaces.plugins.PluginInterface):
if not min_length <= length <= 64:
continue
offset = length.vol["offset"] + DWORD_SIZE_BYTES
passphrase: Bytes = self.context.object(
passphrase: objects.Bytes = self.context.object(
pe_table_name + constants.BANG + "bytes",
layer_name,
offset,
@@ -111,7 +106,7 @@ class Passphrase(interfaces.plugins.PluginInterface):
continue
# TrueCrypt/Common/Password.h::Password struct is padded with
# 3 zero bytes to keep 64-byte alignment.
buf: Bytes = self.context.object(
buf: objects.Bytes = self.context.object(
pe_table_name + constants.BANG + "bytes",
layer_name,
offset + length + 1, # +1 for '\0'-terminated password string
@@ -124,8 +119,8 @@ class Passphrase(interfaces.plugins.PluginInterface):
def _generator(self):
kernel = self.context.modules[self.config["kernel"]]
mods: Iterable[ObjectInterface] = modules.Modules.list_modules(
self.context, self.config["kernel"]
mods: Iterable[interfaces.objects.ObjectInterface] = (
modules.Modules.list_modules(self.context, self.config["kernel"])
)
try:
truecrypt_module_base = next(
volatility3/framework/plugins/windows/vadregexscan.py
+5
View File
@@ -41,6 +41,11 @@ class VadRegExScan(plugins.PluginInterface):
element_type=int,
optional=True,
),
requirements.VersionRequirement(
name="regex_scanner",
component=scanners.RegExScanner,
version=(1, 0, 0),
),
requirements.StringRequirement(
name="pattern", description="RegEx pattern", optional=False
),
volatility3/framework/plugins/windows/vadyarascan.py
+10 -4
View File
@@ -17,8 +17,8 @@ vollog = logging.getLogger(__name__)
class VadYaraScan(interfaces.plugins.PluginInterface):
"""Scans all the Virtual Address Descriptor memory maps using yara."""
_required_framework_version = (2, 4, 0)
_version = (1, 1, 2)
_required_framework_version = (2, 22, 0)
_version = (1, 1, 3)
@classmethod
def get_requirements(cls) -> List[interfaces.configuration.RequirementInterface]:
@@ -93,12 +93,18 @@ class VadYaraScan(interfaces.plugins.PluginInterface):
for offset, rule_name, name, value in scanner(
layer.read(start, size, pad=True), start
):
layer_data = renderers.LayerData(
context=self.context,
offset=offset,
layer_name=layer.name,
length=len(value),
)
yield 0, (
format_hints.Hex(offset),
task.UniqueProcessId,
rule_name,
name,
value,
layer_data,
)
@classmethod
@@ -126,7 +132,7 @@ class VadYaraScan(interfaces.plugins.PluginInterface):
("PID", int),
("Rule", str),
("Component", str),
("Value", bytes),
("Value", renderers.LayerData),
],
self._generator(),
)
volatility3/framework/plugins/windows/verinfo.py
+5
View File
@@ -48,6 +48,11 @@ class VerInfo(interfaces.plugins.PluginInterface):
requirements.VersionRequirement(
name="modules", component=modules.Modules, version=(3, 0, 0)
),
requirements.VersionRequirement(
name="page_start_scanner",
component=scanners.BytesScanner,
version=(1, 0, 0),
),
requirements.BooleanRequirement(
name="extensive",
description="Search physical layer for version information",
volatility3/framework/plugins/yarascan.py
+16 -5
View File
@@ -105,8 +105,8 @@ class YaraScanner(interfaces.layers.ScannerInterface):
class YaraScan(plugins.PluginInterface):
"""Scans kernel memory using yara rules (string or file)."""
_required_framework_version = (2, 0, 0)
_version = (2, 0, 0)
_required_framework_version = (2, 22, 0)
_version = (2, 0, 1)
_yara_x = USE_YARA_X
@classmethod
@@ -118,7 +118,12 @@ class YaraScan(plugins.PluginInterface):
name="primary",
description="Memory layer for the kernel",
architectures=["Intel32", "Intel64"],
)
),
requirements.VersionRequirement(
name="yarascanner",
component=YaraScanner,
version=(2, 1, 1),
),
]
@classmethod
@@ -201,7 +206,13 @@ class YaraScan(plugins.PluginInterface):
for offset, rule_name, name, value in layer.scan(
context=self.context, scanner=YaraScanner(rules=rules)
):
yield 0, (format_hints.Hex(offset), rule_name, name, value)
layer_data = renderers.LayerData(
context=self.context,
offset=offset,
layer_name=layer.name,
length=len(value),
)
yield 0, (format_hints.Hex(offset), rule_name, name, layer_data)
def run(self):
return renderers.TreeGrid(
@@ -209,7 +220,7 @@ class YaraScan(plugins.PluginInterface):
("Offset", format_hints.Hex),
("Rule", str),
("Component", str),
("Value", bytes),
("Value", renderers.LayerData),
],
self._generator(),
)
volatility3/framework/renderers/__init__.py
+77
View File
@@ -8,6 +8,7 @@ or file or graphical output
"""
import collections
import collections.abc
import dataclasses
import datetime
import logging
from typing import Any, Callable, Dict, Iterable, List, Optional, Tuple, TypeVar, Union
@@ -22,16 +23,28 @@ class UnreadableValue(interfaces.renderers.BaseAbsentValue):
"""Class that represents values which are empty because the data cannot be
read."""
def __str__(self) -> str:
"""Fallback method for rendering basic types"""
return "-"
class UnparsableValue(interfaces.renderers.BaseAbsentValue):
"""Class that represents values which are empty because the data cannot be
interpreted correctly."""
def __str__(self) -> str:
"""Fallback method for rendering basic types"""
return "-"
class NotApplicableValue(interfaces.renderers.BaseAbsentValue):
"""Class that represents values which are empty because they don't make
sense for this node."""
def __str__(self) -> str:
"""Fallback method for rendering basic types"""
return "N/A"
class NotAvailableValue(interfaces.renderers.BaseAbsentValue):
"""Class that represents values which cannot be provided now (but might in
@@ -45,6 +58,70 @@ class NotAvailableValue(interfaces.renderers.BaseAbsentValue):
in preference, and only if neither fits should this be used.
"""
def __str__(self) -> str:
"""Fallback method for rendering basic types"""
return "N/A"
##########
### Basic Types
class Disassembly(interfaces.renderers.BasicType):
"""A class to indicate that the bytes provided should be disassembled
(based on the architecture)"""
possible_architectures = ["intel", "intel64", "arm", "arm64"]
def __init__(
self, data: bytes, offset: int = 0, architecture: str = "intel64"
) -> None:
self.data = data
self.architecture = None
if architecture in self.possible_architectures:
self.architecture = architecture
if not isinstance(offset, int):
raise TypeError("Offset must be an integer type")
self.offset = offset
def __str__(self) -> str:
"""Fallback method of rendering"""
return str(self.data)
@dataclasses.dataclass
class LayerData(interfaces.renderers.BasicType):
"""Layer data
This requires the contex to be passed in, in case plugins want to use multiple contexts
and to ensure the TreeGrid interface doesn't change, since this would break all existing plugins
"""
context: "interfaces.context.ContextInterface"
layer_name: str
offset: int
length: int
no_surrounding: bool = False
@staticmethod
def from_object(
object: "interfaces.objects.ObjectInterface",
size: Optional[int] = None,
no_surrounding: bool = True,
):
return LayerData(
context=object._context,
layer_name=object.vol.layer_name,
offset=object.vol.offset,
length=size or object.vol.size,
no_surrounding=no_surrounding,
)
def __str__(self) -> str:
"""Fallback method of rendering"""
data = self.context.layers[self.layer_name].read(self.offset, self.length, True)
return str(data)
class TreeNode(interfaces.renderers.TreeNode):
"""Class representing a particular node in a tree grid."""
volatility3/framework/symbols/linux/network.py
+2 -2
View File
@@ -1,9 +1,9 @@
from volatility3.framework.symbols import intermed
from volatility3.framework.symbols.linux.extensions import network
from volatility3.framework.interfaces.configuration import VersionableInterface
from volatility3.framework.interfaces import configuration
class NetSymbols(VersionableInterface):
class NetSymbols(configuration.VersionableInterface):
_version = (1, 0, 0)
@classmethod
volatility3/framework/symbols/windows/extensions/__init__.py
+4 -3
View File
@@ -18,7 +18,6 @@ from volatility3.framework import (
renderers,
symbols,
)
from volatility3.framework.interfaces.objects import ObjectInterface
from volatility3.framework.layers import intel
from volatility3.framework.objects import utility
from volatility3.framework.renderers import conversion
@@ -413,7 +412,9 @@ class DEVICE_OBJECT(objects.StructType, pool.ExecutiveObject):
header = self.get_object_header()
return header.NameInfo.Name.String # type: ignore
def get_attached_devices(self) -> Generator[ObjectInterface, None, None]:
def get_attached_devices(
self,
) -> Generator[interfaces.objects.ObjectInterface, None, None]:
"""Enumerate the attached device's objects"""
seen = set()
@@ -443,7 +444,7 @@ class DRIVER_OBJECT(objects.StructType, pool.ExecutiveObject):
header = self.get_object_header()
return header.NameInfo.Name.String # type: ignore
def get_devices(self) -> Generator[ObjectInterface, None, None]:
def get_devices(self) -> Generator[interfaces.objects.ObjectInterface, None, None]:
"""Enumerate the driver's device objects"""
seen = set()
volatility3/framework/symbols/windows/extensions/network.py
+3 -5
View File
@@ -4,17 +4,15 @@
import logging
import socket
from typing import Dict, Tuple, List, Union, Optional
from typing import Dict, List, Optional, Tuple, Union
from volatility3.framework import exceptions
from volatility3.framework import objects, interfaces
from volatility3.framework.objects import Array
from volatility3.framework import exceptions, interfaces, objects
from volatility3.framework.renderers import conversion
vollog = logging.getLogger(__name__)
def inet_ntop(address_family: int, packed_ip: Union[List[int], Array]) -> str:
def inet_ntop(address_family: int, packed_ip: Union[List[int], objects.Array]) -> str:
if address_family in [socket.AF_INET6, socket.AF_INET]:
try:
return socket.inet_ntop(address_family, bytes(packed_ip))
volatility3/framework/symbols/windows/extensions/pool.py
+2 -2
View File
@@ -4,7 +4,7 @@ import logging
import struct
from typing import Dict, List, Optional, Tuple, Union
from volatility3.plugins.windows.poolscanner import PoolConstraint
from volatility3.plugins.windows import poolscanner
from volatility3.framework import (
constants,
@@ -28,7 +28,7 @@ class POOL_HEADER(objects.StructType):
def get_object(
self,
constraint: PoolConstraint,
constraint: poolscanner.PoolConstraint,
use_top_down: bool,
kernel_symbol_table: Optional[str] = None,
native_layer_name: Optional[str] = None,
volatility3/framework/symbols/windows/extensions/registry.py
+21 -17
View File
@@ -8,10 +8,7 @@ import struct
from typing import Iterator, Optional, Union, cast
from volatility3.framework import constants, exceptions, interfaces, objects
from volatility3.framework.layers.registry import (
RegistryException,
RegistryHive,
)
from volatility3.framework.layers import registry
vollog = logging.getLogger(__name__)
@@ -102,7 +99,9 @@ class CMHIVE(objects.StructType):
for attr in ["FileFullPath", "FileUserName", "HiveRootPath"]:
with contextlib.suppress(
AttributeError, exceptions.InvalidAddressException, RegistryException
AttributeError,
exceptions.InvalidAddressException,
registry.RegistryException,
):
name = getattr(self, attr)
if name.Length > 0:
@@ -172,7 +171,9 @@ class CM_KEY_NODE(objects.StructType):
Raises TypeError if the key was not instantiated on a RegistryHive layer
"""
if not isinstance(self._context.layers[self.vol.layer_name], RegistryHive):
if not isinstance(
self._context.layers[self.vol.layer_name], registry.RegistryHive
):
raise TypeError("CM_KEY_NODE was not instantiated on a RegistryHive layer")
return bool(self.vol.offset & 0x80000000)
@@ -182,7 +183,7 @@ class CM_KEY_NODE(objects.StructType):
Raises TypeError if the key was not instantiated on a RegistryHive layer
"""
hive = self._context.layers[self.vol.layer_name]
if not isinstance(hive, RegistryHive):
if not isinstance(hive, registry.RegistryHive):
raise TypeError("CM_KEY_NODE was not instantiated on a RegistryHive layer")
for index in range(2):
# Use get_cell because it should *always* be a KeyIndex
@@ -190,7 +191,7 @@ class CM_KEY_NODE(objects.StructType):
yield from self._get_subkeys_recursive(hive, subkey_node)
def _get_subkeys_recursive(
self, hive: RegistryHive, node: interfaces.objects.ObjectInterface
self, hive: "registry.RegistryHive", node: interfaces.objects.ObjectInterface
) -> Iterator["CM_KEY_NODE"]:
"""Recursively descend a node returning subkeys."""
# The keylist appears to include 4 bytes of key name after each value
@@ -200,7 +201,7 @@ class CM_KEY_NODE(objects.StructType):
signature = node.cast("string", max_length=2, encoding="latin-1")
except (
exceptions.InvalidAddressException,
RegistryException,
registry.RegistryException,
):
return None
@@ -229,7 +230,7 @@ class CM_KEY_NODE(objects.StructType):
subnode = hive.get_node(subnode_offset)
except (
exceptions.InvalidAddressException,
RegistryException,
registry.RegistryException,
):
vollog.log(
constants.LOGLEVEL_VVV,
@@ -244,7 +245,7 @@ class CM_KEY_NODE(objects.StructType):
Raises TypeError if the key was not instantiated on a RegistryHive layer
"""
hive = self._context.layers[self.vol.layer_name]
if not isinstance(hive, RegistryHive):
if not isinstance(hive, registry.RegistryHive):
raise TypeError("CM_KEY_NODE was not instantiated on a RegistryHive layer")
try:
@@ -255,7 +256,7 @@ class CM_KEY_NODE(objects.StructType):
if v != 0:
try:
node = hive.get_node(v)
except (RegistryException,) as excp:
except (registry.RegistryException,) as excp:
vollog.debug(f"Invalid address {excp}")
continue
if isinstance(node, CM_KEY_VALUE):
@@ -263,7 +264,7 @@ class CM_KEY_NODE(objects.StructType):
except (
exceptions.InvalidAddressException,
RegistryException,
registry.RegistryException,
) as excp:
vollog.debug(f"Invalid address in get_values iteration: {excp}")
return None
@@ -281,7 +282,7 @@ class CM_KEY_NODE(objects.StructType):
Raises TypeError if the key was not instantiated on a RegistryHive layer
"""
reg = self._context.layers[self.vol.layer_name]
if not isinstance(reg, RegistryHive):
if not isinstance(reg, registry.RegistryHive):
raise TypeError("Key was not instantiated on a RegistryHive layer")
# Using the offset adds a significant delay (since it cannot be cached easily)
# if self.vol.offset == reg.get_node(reg.root_cell_offset).vol.offset:
@@ -320,7 +321,7 @@ class CM_KEY_VALUE(objects.StructType):
data = b""
# Check if the data is stored inline
layer = self._context.layers[self.vol.layer_name]
if not isinstance(layer, RegistryHive):
if not isinstance(layer, registry.RegistryHive):
raise TypeError("Key value was not instantiated on a RegistryHive layer")
# If the high-bit is set
@@ -353,7 +354,10 @@ class CM_KEY_VALUE(objects.StructType):
offset=layer.get_cell(block_offset).vol.offset,
length=amount,
)
except (exceptions.InvalidAddressException, RegistryException):
except (
exceptions.InvalidAddressException,
registry.RegistryException,
):
vollog.debug(
f"Failed to read {amount:x} bytes of data, padding with {amount:x}"
)
@@ -363,7 +367,7 @@ class CM_KEY_VALUE(objects.StructType):
# but the length at the start could be negative so just adding 4 to jump past it
try:
data = layer.read(self.Data + 4, datalen)
except (exceptions.InvalidAddressException, RegistryException):
except (exceptions.InvalidAddressException, registry.RegistryException):
vollog.debug(
f"Failed to read {datalen:x} bytes of data, returning {datalen:x} null bytes"
)
volatility3/framework/symbols/windows/extensions/services.py
+2 -2
View File
@@ -4,7 +4,7 @@
from volatility3.framework import objects, interfaces
from volatility3.framework import exceptions
from volatility3.framework.symbols.wrappers import Flags
from volatility3.framework.symbols import wrappers
from volatility3.framework import renderers
from typing import Union
@@ -91,7 +91,7 @@ class SERVICE_RECORD(objects.StructType):
"SERVICE_INTERACTIVE_PROCESS": 256,
}
type_flags = Flags(choices=SERVICE_TYPE_FLAGS)
type_flags = wrappers.Flags(choices=SERVICE_TYPE_FLAGS)
return "|".join(type_flags(self.Type))
def traverse(self):
volatility3/framework/symbols/windows/pdbutil.py
+1 -2
View File
@@ -16,7 +16,6 @@ from volatility3 import symbols
from volatility3.framework import constants, contexts, exceptions, interfaces
from volatility3.framework.automagic import symbol_cache
from volatility3.framework.configuration import requirements
from volatility3.framework.configuration.requirements import SymbolTableRequirement
from volatility3.framework.symbols import intermed
from volatility3.framework.symbols.windows import pdbconv
@@ -140,7 +139,7 @@ class PDBUtility(interfaces.configuration.VersionableInterface):
requirement_name = interfaces.configuration.path_head(config_path)
# Construct the appropriate symbol table
requirement = SymbolTableRequirement(
requirement = requirements.SymbolTableRequirement(
name=requirement_name, description="PDBUtility generated symbol table"
)
requirement.construct(context, parent_config_path)
volatility3/plugins/windows/registry/certificates.py
+2 -1
View File
@@ -4,6 +4,7 @@ import struct
from typing import Iterator, List, Optional, Tuple, Type
from volatility3.framework import exceptions, interfaces, renderers
from volatility3.framework.layers import registry as registry_layer
from volatility3.framework.configuration import requirements
from volatility3.framework.symbols.windows.extensions import registry
from volatility3.plugins.windows.registry import hivelist, printkey
@@ -80,7 +81,7 @@ class Certificates(interfaces.plugins.PluginInterface):
]:
with contextlib.suppress(
KeyError,
registry.RegistryException,
registry_layer.RegistryException,
exceptions.InvalidAddressException,
):
# Walk it