mirror of
https://github.com/volatilityfoundation/volatility3.git
synced 2026-09-25 11:04:52 +02:00
263 lines
8.5 KiB
Python
263 lines
8.5 KiB
Python
# This file is Copyright 2019 Volatility Foundation and licensed under the Volatility Software License 1.0
|
|
# which is available at https://www.volatilityfoundation.org/license/vsl-v1.0
|
|
#
|
|
from typing import Iterator, Any, Iterable, List, Tuple, Set
|
|
|
|
from volatility3.framework import interfaces, objects, exceptions, constants
|
|
from volatility3.framework.symbols import intermed
|
|
from volatility3.framework.symbols.mac import extensions
|
|
|
|
|
|
class MacKernelIntermedSymbols(intermed.IntermediateSymbolTable):
|
|
provides = {"type": "interface"}
|
|
|
|
def __init__(self, *args, **kwargs) -> None:
|
|
super().__init__(*args, **kwargs)
|
|
|
|
self.set_type_class("proc", extensions.proc)
|
|
self.set_type_class("fileglob", extensions.fileglob)
|
|
self.set_type_class("vnode", extensions.vnode)
|
|
self.set_type_class("vm_map_entry", extensions.vm_map_entry)
|
|
self.set_type_class("vm_map_object", extensions.vm_map_object)
|
|
self.set_type_class("socket", extensions.socket)
|
|
self.set_type_class("inpcb", extensions.inpcb)
|
|
self.set_type_class("ifnet", extensions.ifnet)
|
|
self.set_type_class("sockaddr_dl", extensions.sockaddr_dl)
|
|
self.set_type_class("sockaddr", extensions.sockaddr)
|
|
self.set_type_class("sysctl_oid", extensions.sysctl_oid)
|
|
self.set_type_class("kauth_scope", extensions.kauth_scope)
|
|
# https://developer.apple.com/documentation/kernel/queue_head_t
|
|
self.set_type_class("queue_entry", extensions.queue_entry)
|
|
self.optional_set_type_class("queue_head_t", extensions.queue_entry)
|
|
|
|
|
|
class MacUtilities(interfaces.configuration.VersionableInterface):
|
|
"""Class with multiple useful mac functions."""
|
|
|
|
"""
|
|
Version History:
|
|
1.1.0 -> added walk_list_head API
|
|
1.2.0 -> added walk_slist API
|
|
1.3.0 -> add parameter to lookup_module_address to pass kernel module name
|
|
"""
|
|
_version = (1, 3, 0)
|
|
_required_framework_version = (2, 0, 0)
|
|
|
|
@classmethod
|
|
def mask_mods_list(
|
|
cls,
|
|
context: interfaces.context.ContextInterface,
|
|
layer_name: str,
|
|
mods: Iterator[Any],
|
|
) -> List[Tuple[interfaces.objects.ObjectInterface, Any, Any]]:
|
|
"""
|
|
A helper function to mask the starting and end address of kernel modules
|
|
"""
|
|
mask = context.layers[layer_name].address_mask
|
|
|
|
return [
|
|
(
|
|
objects.utility.array_to_string(mod.name),
|
|
mod.address & mask,
|
|
(mod.address & mask) + mod.size,
|
|
)
|
|
for mod in mods
|
|
]
|
|
|
|
@classmethod
|
|
def generate_kernel_handler_info(
|
|
cls,
|
|
context: interfaces.context.ContextInterface,
|
|
layer_name: str,
|
|
kernel, # ikelos - how to type this??
|
|
mods_list: Iterator[Any],
|
|
):
|
|
try:
|
|
start_addr = kernel.object_from_symbol("vm_kernel_stext")
|
|
except exceptions.SymbolError:
|
|
start_addr = kernel.object_from_symbol("stext")
|
|
|
|
try:
|
|
end_addr = kernel.object_from_symbol("vm_kernel_etext")
|
|
except exceptions.SymbolError:
|
|
end_addr = kernel.object_from_symbol("etext")
|
|
|
|
mask = context.layers[layer_name].address_mask
|
|
|
|
start_addr = start_addr & mask
|
|
end_addr = end_addr & mask
|
|
|
|
return [("__kernel__", start_addr, end_addr)] + MacUtilities.mask_mods_list(
|
|
context, layer_name, mods_list
|
|
)
|
|
|
|
@classmethod
|
|
def lookup_module_address(
|
|
cls,
|
|
context: interfaces.context.ContextInterface,
|
|
handlers: Iterator[Any],
|
|
target_address,
|
|
kernel_module_name: str = None,
|
|
):
|
|
mod_name = "UNKNOWN"
|
|
symbol_name = "N/A"
|
|
|
|
module_shift = 0
|
|
if kernel_module_name:
|
|
module = context.modules[kernel_module_name]
|
|
module_shift = module.offset
|
|
|
|
for name, start, end in handlers:
|
|
if start <= target_address <= end:
|
|
mod_name = name
|
|
if name == "__kernel__":
|
|
symbols = list(
|
|
context.symbol_space.get_symbols_by_location(
|
|
target_address - module_shift
|
|
)
|
|
)
|
|
|
|
if len(symbols) > 0:
|
|
symbol_name = (
|
|
str(symbols[0].split(constants.BANG)[1])
|
|
if constants.BANG in symbols[0]
|
|
else str(symbols[0])
|
|
)
|
|
|
|
break
|
|
|
|
return mod_name, symbol_name
|
|
|
|
@classmethod
|
|
def files_descriptors_for_process(
|
|
cls,
|
|
context: interfaces.context.ContextInterface,
|
|
symbol_table_name: str,
|
|
task: interfaces.objects.ObjectInterface,
|
|
):
|
|
"""Creates a generator for the file descriptors of a process
|
|
|
|
Args:
|
|
symbol_table_name: The name of the symbol table associated with the process
|
|
context:
|
|
task: The process structure to enumerate file descriptors from
|
|
|
|
Return:
|
|
A 3 element tuple is yielded for each file descriptor:
|
|
1) The file's object
|
|
2) The path referenced by the descriptor.
|
|
The path is either empty, the full path of the file in the file system, or the formatted name for sockets, pipes, etc.
|
|
3) The file descriptor number
|
|
"""
|
|
|
|
try:
|
|
num_fds = task.p_fd.fd_lastfile
|
|
except exceptions.InvalidAddressException:
|
|
num_fds = 1024
|
|
|
|
try:
|
|
nfiles = task.p_fd.fd_nfiles
|
|
except exceptions.InvalidAddressException:
|
|
nfiles = 1024
|
|
|
|
if nfiles > num_fds:
|
|
num_fds = nfiles
|
|
|
|
if num_fds > 4096:
|
|
num_fds = 1024
|
|
|
|
file_type = symbol_table_name + constants.BANG + "fileproc"
|
|
|
|
try:
|
|
table_addr = task.p_fd.fd_ofiles.dereference()
|
|
except exceptions.InvalidAddressException:
|
|
return None
|
|
|
|
fds = objects.utility.array_of_pointers(
|
|
table_addr, count=num_fds, subtype=file_type, context=context
|
|
)
|
|
|
|
for fd_num, f in enumerate(fds):
|
|
if f != 0:
|
|
try:
|
|
ftype = f.f_fglob.get_fg_type()
|
|
except exceptions.InvalidAddressException:
|
|
continue
|
|
|
|
if ftype == "VNODE":
|
|
vnode = f.f_fglob.fg_data.dereference().cast("vnode")
|
|
path = vnode.full_path()
|
|
elif ftype:
|
|
path = f"<{ftype.lower()}>"
|
|
|
|
yield f, path, fd_num
|
|
|
|
@classmethod
|
|
def _walk_iterable(
|
|
cls,
|
|
queue: interfaces.objects.ObjectInterface,
|
|
list_head_member: str,
|
|
list_next_member: str,
|
|
next_member: str,
|
|
max_elements: int = 4096,
|
|
) -> Iterable[interfaces.objects.ObjectInterface]:
|
|
seen: Set[int] = set()
|
|
|
|
try:
|
|
current = queue.member(attr=list_head_member)
|
|
except exceptions.InvalidAddressException:
|
|
return None
|
|
|
|
while current:
|
|
if current.vol.offset in seen:
|
|
break
|
|
|
|
seen.add(current.vol.offset)
|
|
|
|
if len(seen) == max_elements:
|
|
break
|
|
|
|
if current.is_readable():
|
|
yield current
|
|
|
|
try:
|
|
current = current.member(attr=next_member).member(attr=list_next_member)
|
|
except exceptions.InvalidAddressException:
|
|
break
|
|
|
|
@classmethod
|
|
def walk_tailq(
|
|
cls,
|
|
queue: interfaces.objects.ObjectInterface,
|
|
next_member: str,
|
|
max_elements: int = 4096,
|
|
) -> Iterable[interfaces.objects.ObjectInterface]:
|
|
for element in cls._walk_iterable(
|
|
queue, "tqh_first", "tqe_next", next_member, max_elements
|
|
):
|
|
yield element
|
|
|
|
@classmethod
|
|
def walk_list_head(
|
|
cls,
|
|
queue: interfaces.objects.ObjectInterface,
|
|
next_member: str,
|
|
max_elements: int = 4096,
|
|
) -> Iterable[interfaces.objects.ObjectInterface]:
|
|
for element in cls._walk_iterable(
|
|
queue, "lh_first", "le_next", next_member, max_elements
|
|
):
|
|
yield element
|
|
|
|
@classmethod
|
|
def walk_slist(
|
|
cls,
|
|
queue: interfaces.objects.ObjectInterface,
|
|
next_member: str,
|
|
max_elements: int = 4096,
|
|
) -> Iterable[interfaces.objects.ObjectInterface]:
|
|
for element in cls._walk_iterable(
|
|
queue, "slh_first", "sle_next", next_member, max_elements
|
|
):
|
|
yield element
|