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

Merge pull request #1110 from dgmcdona/dgmcdona/windows-callbacks

Browse Source 
Windows: Add support for missing callback types
This commit is contained in:
ikelos
2024-07-04 23:04:55 +01:00
committed by GitHub
parent 1e58547aa9 4d6f4bba0d
commit 4f69638d4d
6 changed files with 712 additions and 52 deletions
Download Patch File Download Diff File Expand all files Collapse all files
volatility3/framework/plugins/windows/callbacks.py
+317 -30
View File
@@ -2,16 +2,24 @@
# which is available at https://www.volatilityfoundation.org/license/vsl-v1.0
#
import logging
import contextlib
from typing import List, Iterable, Tuple, Optional, Union
import logging
from typing import Dict, Iterable, List, Optional, Tuple, Union, cast
from volatility3.framework import constants, exceptions, renderers, interfaces, symbols
from volatility3.framework import (
constants,
exceptions,
interfaces,
objects,
renderers,
symbols,
)
from volatility3.framework.configuration import requirements
from volatility3.framework.renderers import format_hints
from volatility3.framework.symbols import intermed
from volatility3.framework.symbols.windows import versions
from volatility3.plugins.windows import ssdt
from volatility3.framework.symbols.windows.extensions import callbacks
from volatility3.plugins.windows import driverirp, handles, poolscanner, ssdt
vollog = logging.getLogger(__name__)
@@ -20,7 +28,7 @@ class Callbacks(interfaces.plugins.PluginInterface):
"""Lists kernel callbacks and notification routines."""
_required_framework_version = (2, 0, 0)
_version = (1, 0, 0)
_version = (2, 0, 0)
@classmethod
def get_requirements(cls) -> List[interfaces.configuration.RequirementInterface]:
@@ -33,27 +41,154 @@ class Callbacks(interfaces.plugins.PluginInterface):
requirements.PluginRequirement(
name="ssdt", plugin=ssdt.SSDT, version=(1, 0, 0)
),
requirements.PluginRequirement(
name="poolscanner", plugin=poolscanner.PoolScanner, version=(1, 0, 0)
),
requirements.PluginRequirement(
name="driverirp", plugin=driverirp.DriverIrp, version=(1, 0, 0)
),
requirements.PluginRequirement(
name="handles", plugin=handles.Handles, version=(1, 0, 0)
),
]
@staticmethod
def create_callback_table(
@classmethod
def create_callback_scan_constraints(
cls,
context: interfaces.context.ContextInterface,
symbol_table: str,
config_path: str,
) -> str:
"""Creates a symbol table for a set of callbacks.
is_vista_or_above: bool,
) -> List[poolscanner.PoolConstraint]:
"""Creates a list of Pool Tag Constraints for callback objects.
Args:
context: The context to retrieve required elements (layers, symbol tables) from
symbol_table: The name of an existing symbol table containing the kernel symbols
config_path: The configuration path within the context of the symbol table to create
symbol_table: The name of an existing symbol table containing the symbols / types
is_vista_or_above: A boolean indicating whether the OS version is Vista or newer.
Returns:
The name of the constructed callback table
The list containing the built constraints.
"""
native_types = context.symbol_space[symbol_table].natives
is_64bit = symbols.symbol_table_is_64bit(context, symbol_table)
table_mapping = {"nt_symbols": symbol_table}
constraints = cls._create_default_scan_constraints(context, symbol_table)
if is_vista_or_above:
constraints += cls._create_scan_constraints_vista_and_above(symbol_table)
return constraints
@staticmethod
def _create_default_scan_constraints(
context: interfaces.context.ContextInterface, symbol_table: str
) -> List[poolscanner.PoolConstraint]:
shutdown_packet_size = context.symbol_space.get_type(
symbol_table + constants.BANG + "_SHUTDOWN_PACKET"
).size
generic_callback_size = context.symbol_space.get_type(
symbol_table + constants.BANG + "_GENERIC_CALLBACK"
).size
notification_packet_size = context.symbol_space.get_type(
symbol_table + constants.BANG + "_NOTIFICATION_PACKET"
).size
return [
poolscanner.PoolConstraint(
b"IoFs",
type_name=symbol_table + constants.BANG + "_NOTIFICATION_PACKET",
size=(notification_packet_size, None),
page_type=poolscanner.PoolType.NONPAGED
| poolscanner.PoolType.PAGED
| poolscanner.PoolType.FREE,
),
poolscanner.PoolConstraint(
b"IoSh",
type_name=symbol_table + constants.BANG + "_SHUTDOWN_PACKET",
size=(shutdown_packet_size, None),
page_type=poolscanner.PoolType.NONPAGED
| poolscanner.PoolType.PAGED
| poolscanner.PoolType.FREE,
index=(0, 0),
),
poolscanner.PoolConstraint(
b"Cbrb",
type_name=symbol_table + constants.BANG + "_GENERIC_CALLBACK",
size=(generic_callback_size, None),
page_type=poolscanner.PoolType.NONPAGED
| poolscanner.PoolType.PAGED
| poolscanner.PoolType.FREE,
),
]
@staticmethod
def _create_scan_constraints_vista_and_above(
symbol_table: str,
) -> List[poolscanner.PoolConstraint]:
"""Creates a list of Pool Tag Constraints for callback objects.
Args:
context: The context to retrieve required elements (layers, symbol tables) from
symbol_table: The name of an existing symbol table containing the symbols / types
Returns:
The list containing the built constraints.
"""
return [
poolscanner.PoolConstraint(
b"DbCb",
type_name=symbol_table + constants.BANG + "_DBGPRINT_CALLBACK",
size=(0x20, 0x40),
page_type=poolscanner.PoolType.NONPAGED
| poolscanner.PoolType.PAGED
| poolscanner.PoolType.FREE,
),
poolscanner.PoolConstraint(
b"Pnp9",
type_name=symbol_table + constants.BANG + "_NOTIFY_ENTRY_HEADER",
size=(0x30, None),
page_type=poolscanner.PoolType.NONPAGED
| poolscanner.PoolType.PAGED
| poolscanner.PoolType.FREE,
index=(1, 1),
),
poolscanner.PoolConstraint(
b"PnpD",
type_name=symbol_table + constants.BANG + "_NOTIFY_ENTRY_HEADER",
size=(0x40, None),
page_type=poolscanner.PoolType.NONPAGED
| poolscanner.PoolType.PAGED
| poolscanner.PoolType.FREE,
index=(1, 1),
),
poolscanner.PoolConstraint(
b"PnpC",
type_name=symbol_table + constants.BANG + "_NOTIFY_ENTRY_HEADER",
size=(0x38, None),
page_type=poolscanner.PoolType.NONPAGED
| poolscanner.PoolType.PAGED
| poolscanner.PoolType.FREE,
index=(1, 1),
),
]
@classmethod
def create_callback_symbol_table(
cls,
context: interfaces.context.ContextInterface,
nt_symbol_table: str,
config_path: str,
) -> str:
"""Creates a symbol table for kernel callback objects.
Args:
context: The context to retrieve required elements (layers, symbol tables) from
nt_symbol_table: The name of the table containing the kernel symbols
config_path: The config path where to find symbol files
Returns:
The name of the constructed symbol table
"""
native_types = context.symbol_space[nt_symbol_table].natives
is_64bit = symbols.symbol_table_is_64bit(context, nt_symbol_table)
table_mapping = {"nt_symbols": nt_symbol_table}
if is_64bit:
symbol_filename = "callbacks-x64"
@@ -65,10 +200,143 @@ class Callbacks(interfaces.plugins.PluginInterface):
config_path,
"windows",
symbol_filename,
class_types=callbacks.class_types_x86 if not is_64bit else None,
native_types=native_types,
table_mapping=table_mapping,
)
@classmethod
def scan(
cls,
context: interfaces.context.ContextInterface,
layer_name: str,
nt_symbol_table: str,
callback_symbol_table: str,
) -> Iterable[
Tuple[
Union[str, interfaces.renderers.BaseAbsentValue],
int,
Union[str, interfaces.renderers.BaseAbsentValue],
]
]:
"""Scans for callback objects using the poolscanner module and constraints.
Args:
context: The context to retrieve required elements (layers, symbol tables) from
layer_name: The name of the layer on which to operate
nt_symbol_table: The name of the table containing the kernel symbols
callback_symbol_table: The name of the table containing the callback object symbols (_SHUTDOWN_PACKET etc.)
Returns:
A list of callback objects found by scanning the `layer_name` layer for callback pool signatures
"""
is_vista_or_later = versions.is_vista_or_later(
context=context, symbol_table=nt_symbol_table
)
type_map = handles.Handles.get_type_map(context, layer_name, nt_symbol_table)
constraints = cls.create_callback_scan_constraints(
context, callback_symbol_table, is_vista_or_later
)
for (
_constraint,
mem_object,
_header,
) in poolscanner.PoolScanner.generate_pool_scan(
context, layer_name, nt_symbol_table, constraints
):
try:
if hasattr(mem_object, "is_valid") and not mem_object.is_valid():
continue
yield cls._process_scanned_callback(mem_object, type_map)
except exceptions.InvalidAddressException:
continue
@classmethod
def _process_scanned_callback(
cls, memory_object: objects.StructType, type_map: Dict[int, str]
) -> Tuple[
Union[str, interfaces.renderers.BaseAbsentValue],
int,
Union[str, interfaces.renderers.BaseAbsentValue],
]:
symbol_table = memory_object.get_symbol_table_name()
type_name = memory_object.vol.type_name
if isinstance(memory_object, callbacks._SHUTDOWN_PACKET) or (
type_name == symbol_table + constants.BANG + "_SHUTDOWN_PACKET"
):
callback_type = "IoRegisterShutdownNotification"
try:
driver = memory_object.DeviceObject.DriverObject
index = driverirp.MAJOR_FUNCTIONS.index("IRP_MJ_SHUTDOWN")
callback_address = driver.MajorFunction[index]
details = driver.DriverName.String or renderers.UnparsableValue()
except exceptions.InvalidAddressException:
callback_address = memory_object.vol.offset
details = renderers.NotApplicableValue()
elif type_name == symbol_table + constants.BANG + "_NOTIFICATION_PACKET":
callback_type = "IoRegisterFsRegistrationChange"
callback_address = memory_object.NotificationRoutine
details = renderers.NotApplicableValue()
elif type_name == symbol_table + constants.BANG + "_NOTIFY_ENTRY_HEADER":
driver = (
memory_object.DriverObject.dereference()
if memory_object.DriverObject.is_readable()
else None
)
if driver:
# Instantiate an object header for the driver name
header = driver.get_object_header()
try:
if header.get_object_type(type_map) == "Driver":
# Grab the object name
details = header.NameInfo.Name.String
else:
details = renderers.NotApplicableValue()
except exceptions.InvalidAddressException:
details = renderers.UnreadableValue()
else:
details = renderers.UnreadableValue()
callback_type = (
memory_object.EventCategory.description
if memory_object.EventCategory.is_valid_choice
else renderers.UnparsableValue()
)
callback_address = memory_object.CallbackRoutine
elif type_name == symbol_table + constants.BANG + "_GENERIC_CALLBACK":
callback_type = "GenericKernelCallback"
callback_address = memory_object.Callback
details = renderers.NotApplicableValue()
elif type_name == symbol_table + constants.BANG + "_DBGPRINT_CALLBACK":
callback_type = "DbgSetDebugPrintCallback"
callback_address = memory_object.Function
details = renderers.NotApplicableValue()
else:
raise ValueError(f"Unexpected object type {type_name}")
return (
callback_type,
callback_address,
(
details
if not isinstance(details, interfaces.renderers.BaseAbsentValue)
else details
),
)
@classmethod
def list_notify_routines(
cls,
@@ -115,11 +383,14 @@ class Callbacks(interfaces.plugins.PluginInterface):
else:
count = 8
fast_refs = ntkrnlmp.object(
object_type="array",
offset=symbol_offset,
subtype=ntkrnlmp.get_type("_EX_FAST_REF"),
count=count,
fast_refs = cast(
List[objects.Pointer],
ntkrnlmp.object(
object_type="array",
offset=symbol_offset,
subtype=ntkrnlmp.get_type("_EX_FAST_REF"),
count=count,
),
)
for fast_ref in fast_refs:
@@ -159,11 +430,14 @@ class Callbacks(interfaces.plugins.PluginInterface):
if callback_count == 0:
return None
fast_refs = ntkrnlmp.object(
object_type="array",
offset=symbol_offset,
subtype=ntkrnlmp.get_type("_EX_FAST_REF"),
count=callback_count,
fast_refs = cast(
List[objects.Pointer],
ntkrnlmp.object(
object_type="array",
offset=symbol_offset,
subtype=ntkrnlmp.get_type("_EX_FAST_REF"),
count=callback_count,
),
)
for fast_ref in fast_refs:
@@ -265,7 +539,13 @@ class Callbacks(interfaces.plugins.PluginInterface):
layer_name: str,
symbol_table: str,
callback_table_name: str,
) -> Iterable[Tuple[str, int, str]]:
) -> Iterable[
Tuple[
str,
int,
interfaces.renderers.BaseAbsentValue,
]
]:
"""Lists all kernel bugcheck reason callbacks.
Args:
@@ -323,7 +603,13 @@ class Callbacks(interfaces.plugins.PluginInterface):
layer_name: str,
symbol_table: str,
callback_table_name: str,
) -> Iterable[Tuple[str, int, str]]:
) -> Iterable[
Tuple[
str,
int,
Union[interfaces.objects.ObjectInterface, renderers.UnreadableValue],
]
]:
"""Lists all kernel bugcheck callbacks.
Args:
@@ -372,7 +658,7 @@ class Callbacks(interfaces.plugins.PluginInterface):
def _generator(self):
kernel = self.context.modules[self.config["kernel"]]
callback_table_name = self.create_callback_table(
callback_symbol_table = self.create_callback_symbol_table(
self.context, kernel.symbol_table_name, self.config_path
)
@@ -385,6 +671,7 @@ class Callbacks(interfaces.plugins.PluginInterface):
self.list_bugcheck_callbacks,
self.list_bugcheck_reason_callbacks,
self.list_registry_callbacks,
self.scan,
)
for callback_method in callback_methods:
@@ -392,7 +679,7 @@ class Callbacks(interfaces.plugins.PluginInterface):
self.context,
kernel.layer_name,
kernel.symbol_table_name,
callback_table_name,
callback_symbol_table,
):
if callback_detail is None:
detail = renderers.NotApplicableValue()
volatility3/framework/plugins/windows/driverirp.py
+1
View File
@@ -44,6 +44,7 @@ class DriverIrp(interfaces.plugins.PluginInterface):
"""List IRPs for drivers in a particular windows memory image."""
_required_framework_version = (2, 0, 0)
_version = (1, 0, 0)
@classmethod
def get_requirements(cls):
volatility3/framework/symbols/intermed.py
+1 -1
View File
@@ -282,7 +282,7 @@ class IntermediateSymbolTable(interfaces.symbols.SymbolTableInterface):
urls = list(cls.file_symbol_url(sub_path, filename))
if not urls:
raise FileNotFoundError(
"No symbol files found at provided filename: {}", filename
f"No symbol files found at provided filename: {filename}",
)
table_name = context.symbol_space.free_table_name(filename)
table = cls(
volatility3/framework/symbols/windows/callbacks-x64.json
+169 -15
View File
@@ -1,6 +1,18 @@
{
"symbols": {},
"enums": {},
"enums": {
"EventCategory": {
"base": "long",
"constants": {
"EventCategoryReserved": 0,
"EventCategoryHardwareProfileChange": 1,
"EventCategoryDeviceInterfaceChange": 2,
"EventCategoryTargetDeviceChange": 3,
"EventCategoryKernelSoftRestart": 4
},
"size": 4
}
},
"base_types": {
"unsigned long": {
"kind": "int",
@@ -48,19 +60,161 @@
"user_types": {
"_GENERIC_CALLBACK": {
"fields": {
"Callback": {
"type": {
"kind": "pointer",
"subtype": {
"kind": "base",
"name": "void"
}
"Callback": {
"type": {
"kind": "pointer",
"subtype": {
"kind": "base",
"name": "void"
}
},
"offset": 8
},
"offset": 8
}
"Associated": {
"type": {
"kind": "pointer",
"subtype": {
"kind": "base",
"name": "void"
}
},
"offset": 16
}
},
"kind": "struct",
"size": 16
"size": 24
},
"_NOTIFICATION_PACKET": {
"fields": {
"ListEntry": {
"type": {
"kind": "struct",
"name": "nt_symbols!_LIST_ENTRY"
},
"offset": 0
},
"DriverObject": {
"type": {
"kind": "pointer",
"subtype": {
"kind": "struct",
"name": "nt_symbols!_DRIVER_OBJECT"
}
},
"offset": 16
},
"NotificationRoutine": {
"type": {
"kind": "base",
"name": "unsigned int"
},
"offset": 24
}
},
"kind": "struct",
"size": 48
},
"_SHUTDOWN_PACKET": {
"fields": {
"Entry": {
"type": {
"kind": "struct",
"name": "nt_symbols!_LIST_ENTRY"
},
"offset": 0
},
"DeviceObject": {
"type": {
"kind": "pointer",
"subtype": {
"kind": "struct",
"name": "nt_symbols!_DEVICE_OBJECT"
}
},
"offset": 16
}
},
"kind": "struct",
"size": 24
},
"_DBGPRINT_CALLBACK": {
"fields": {
"Function": {
"type": {
"kind": "pointer",
"subtype": {
"kind": "base",
"name": "void"
}
},
"offset": 16
}
},
"kind": "struct",
"size": 24
},
"_NOTIFY_ENTRY_HEADER": {
"fields": {
"ListEntry": {
"type": {
"kind": "struct",
"name": "nt_symbols!_LIST_ENTRY"
},
"offset": 0
},
"EventCategory": {
"type": {
"kind": "enum",
"name": "EventCategory"
},
"offset": 16
},
"CallbackRoutine": {
"type": {
"kind": "pointer",
"subtype": {
"kind": "base",
"name": "void"
}
},
"offset": 32
},
"DriverObject": {
"type": {
"kind": "pointer",
"subtype": {
"kind": "struct",
"name": "nt_symbols!_DRIVER_OBJECT"
}
},
"offset": 48
}
},
"kind": "struct",
"size": 64
},
"_REGISTRY_CALLBACK": {
"fields": {
"ListEntry": {
"type": {
"kind": "struct",
"name": "nt_symbols!_LIST_ENTRY"
},
"offset": 0
},
"Function": {
"type": {
"kind": "pointer",
"subtype": {
"kind": "base",
"name": "void"
}
},
"offset": 40
}
},
"kind": "struct",
"size": 48
},
"_KBUGCHECK_CALLBACK_RECORD": {
"fields": {
@@ -184,10 +338,10 @@
},
"metadata": {
"producer": {
"version": "0.0.1",
"name": "mhl by hand",
"datetime": "2019-08-27T18:17:16.417006"
"version": "0.0.2",
"name": "dgmcdona by hand",
"datetime": "2024-02-02T19:32:00.000000"
},
"format": "4.0.0"
}
}
}
volatility3/framework/symbols/windows/callbacks-x86.json
+173 -6
View File
@@ -1,6 +1,18 @@
{
"symbols": {},
"enums": {},
"enums": {
"EventCategory": {
"base": "long",
"constants": {
"EventCategoryReserved": 0,
"EventCategoryHardwareProfileChange": 1,
"EventCategoryDeviceInterfaceChange": 2,
"EventCategoryTargetDeviceChange": 3,
"EventCategoryKernelSoftRestart": 4
},
"size": 4
}
},
"base_types": {
"unsigned long": {
"kind": "int",
@@ -46,6 +58,151 @@
}
},
"user_types": {
"_NOTIFICATION_PACKET": {
"fields": {
"ListEntry": {
"type": {
"kind": "struct",
"name": "nt_symbols!_LIST_ENTRY"
},
"offset": 0
},
"DriverObject": {
"type": {
"kind": "pointer",
"subtype": {
"kind": "struct",
"name": "nt_symbols!_DRIVER_OBJECT"
}
},
"offset": 8
},
"NotificationRoutine": {
"type": {
"kind": "base",
"name": "unsigned int"
},
"offset": 12
}
},
"kind": "struct",
"size": 16
},
"_SHUTDOWN_PACKET": {
"fields": {
"Entry": {
"type": {
"kind": "struct",
"name": "nt_symbols!_LIST_ENTRY"
},
"offset": 0
},
"DeviceObject": {
"type": {
"kind": "pointer",
"subtype": {
"kind": "struct",
"name": "nt_symbols!_DEVICE_OBJECT"
}
},
"offset": 8
}
},
"kind": "struct",
"size": 12
},
"_REGISTRY_CALLBACK": {
"fields": {
"ListEntry": {
"type": {
"kind": "struct",
"name": "nt_symbols!_LIST_ENTRY"
},
"offset": 0
},
"Function": {
"type": {
"kind": "pointer",
"subtype": {
"kind": "base",
"name": "void"
}
},
"offset": 28
}
},
"kind": "struct",
"size": 32
},
"_REGISTRY_CALLBACK_LEGACY": {
"fields": {
"CreateTime": {
"type": {
"kind": "pointer",
"subtype": {
"kind": "base",
"name": "void"
}
},
"offset": 28
}
},
"kind": "struct",
"size": 32
},
"_DBGPRINT_CALLBACK": {
"fields": {
"Function": {
"type": {
"kind": "pointer",
"subtype": {
"kind": "base",
"name": "void"
}
},
"offset": 8
}
},
"kind": "struct",
"size": 20
},
"_NOTIFY_ENTRY_HEADER": {
"fields": {
"ListEntry": {
"type": {
"kind": "struct",
"name": "nt_symbols!_LIST_ENTRY"
},
"offset": 0
},
"EventCategory": {
"type": {
"kind": "enum",
"name": "EventCategory"
},
"offset": 8
},
"CallbackRoutine": {
"type": {
"kind": "base",
"name": "unsigned int"
},
"offset": 20
},
"DriverObject": {
"type": {
"kind": "pointer",
"subtype": {
"kind": "struct",
"name": "nt_symbols!_DRIVER_OBJECT"
}
},
"offset": 28
}
},
"kind": "struct",
"size": 32
},
"_GENERIC_CALLBACK": {
"fields": {
"Callback": {
@@ -57,10 +214,20 @@
}
},
"offset": 4
},
"Associated": {
"type": {
"kind": "pointer",
"subtype": {
"kind": "base",
"name": "void"
}
},
"offset": 8
}
},
"kind": "struct",
"size": 8
"size": 12
},
"_KBUGCHECK_CALLBACK_RECORD": {
"fields": {
@@ -184,10 +351,10 @@
},
"metadata": {
"producer": {
"version": "0.0.1",
"name": "mhl by hand",
"datetime": "2019-08-27T18:17:16.417006"
"version": "0.0.2",
"name": "dgmcdona by hand",
"datetime": "2024-02-02T19:32:00.000000"
},
"format": "4.0.0"
}
}
}
volatility3/framework/symbols/windows/extensions/callbacks.py
+51
View File
@@ -0,0 +1,51 @@
import logging
from volatility3.framework import exceptions, objects
from volatility3.framework.symbols.windows.extensions import pool
vollog = logging.getLogger(__name__)
class _SHUTDOWN_PACKET(objects.StructType, pool.ExecutiveObject):
"""Class for _SHUTDOWN_PACKET objects found in IoSh pools.
This class serves as a base class for all pooled shutdown callback packets.
It exposes a function which sanity-checks structure members.
"""
def is_valid(self) -> bool:
"""
Perform some checks.
"""
try:
if not (
self.Entry.Flink.is_readable()
and self.Entry.Blink.is_readable()
and self.DeviceObject.is_readable()
):
return False
device = self.DeviceObject
if not device or not (device.DriverObject.DriverStart % 0x1000 == 0):
vollog.debug(
f"callback obj 0x{self.vol.offset:x} invalid due to invalid device object"
)
return False
except exceptions.InvalidAddressException:
vollog.debug(
f"callback obj 0x{self.vol.offset:x} invalid due to invalid address access"
)
return False
try:
header = device.get_object_header()
valid = header.NameInfo.Name == "Device"
return valid
except ValueError:
vollog.debug(f"Could not get NameInfo for object at 0x{self.vol.offset:x}")
return False
class_types_x86 = {"_SHUTDOWN_PACKET": _SHUTDOWN_PACKET}