Merge branch 'develop' into issues/issue1577

This commit is contained in:
ikelos
2025-01-30 00:34:41 +00:00
committed by GitHub
19 changed files with 152 additions and 110 deletions
+5 -2
View File
@@ -33,11 +33,12 @@ def array_to_string(
) -> interfaces.objects.ObjectInterface:
"""Takes a volatility Array of characters and returns a string."""
# TODO: Consider checking the Array's target is a native char
if count is None:
count = array.vol.count
if not isinstance(array, objects.Array):
raise TypeError("Array_to_string takes an Array of char")
if count is None:
count = array.vol.count
return array.cast("string", max_length=count, errors=errors)
@@ -45,8 +46,10 @@ def pointer_to_string(pointer: "objects.Pointer", count: int, errors: str = "rep
"""Takes a volatility Pointer to characters and returns a string."""
if not isinstance(pointer, objects.Pointer):
raise TypeError("pointer_to_string takes a Pointer")
if count < 1:
raise ValueError("pointer_to_string requires a positive count")
char = pointer.dereference()
return char.cast("string", max_length=count, errors=errors)
@@ -18,7 +18,7 @@ class Envars(plugins.PluginInterface):
"""Lists processes with their environment variables"""
_required_framework_version = (2, 13, 0)
_version = (2, 0, 0)
_version = (2, 0, 1)
@classmethod
def get_requirements(cls):
@@ -40,8 +40,9 @@ class Envars(plugins.PluginInterface):
),
]
@staticmethod
@classmethod
def get_task_env_variables(
cls,
context: interfaces.context.ContextInterface,
task: interfaces.objects.ObjectInterface,
env_area_max_size: int = 8192,
@@ -16,8 +16,7 @@ class Hidden_modules(interfaces.plugins.PluginInterface):
"""Carves memory to find hidden kernel modules"""
_required_framework_version = (2, 10, 0)
_version = (1, 0, 0)
_version = (1, 0, 1)
@classmethod
def get_requirements(cls) -> List[interfaces.configuration.RequirementInterface]:
@@ -32,8 +31,9 @@ class Hidden_modules(interfaces.plugins.PluginInterface):
),
]
@staticmethod
@classmethod
def get_modules_memory_boundaries(
cls,
context: interfaces.context.ContextInterface,
vmlinux_module_name: str,
) -> Tuple[int]:
@@ -36,7 +36,7 @@ class MountInfo(plugins.PluginInterface):
"""Lists mount points on processes mount namespaces"""
_required_framework_version = (2, 2, 0)
_version = (1, 2, 3)
_version = (1, 2, 4)
@classmethod
def get_requirements(cls) -> List[interfaces.configuration.RequirementInterface]:
@@ -152,9 +152,11 @@ class MountInfo(plugins.PluginInterface):
if not (
task
and task.fs
and task.fs.root
and task.fs.is_readable()
and task.nsproxy
and task.nsproxy.is_readable()
and task.nsproxy.mnt_ns
and task.nsproxy.mnt_ns.is_readable()
):
# This task doesn't have all the information required.
# It should be a kernel < 2.6.30
@@ -104,7 +104,7 @@ class Files(plugins.PluginInterface, timeliner.TimeLinerInterface):
_required_framework_version = (2, 0, 0)
_version = (1, 0, 2)
_version = (1, 0, 3)
@classmethod
def get_requirements(cls) -> List[interfaces.configuration.RequirementInterface]:
@@ -360,8 +360,8 @@ class Files(plugins.PluginInterface, timeliner.TimeLinerInterface):
yield description, timeliner.TimeLinerType.MODIFIED, inode_out.modification_time
yield description, timeliner.TimeLinerType.CHANGED, inode_out.change_time
@staticmethod
def format_fields_with_headers(headers, generator):
@classmethod
def format_fields_with_headers(cls, headers, generator):
"""Uses the headers type to cast the fields obtained from the generator"""
for level, fields in generator:
formatted_fields = []
@@ -405,7 +405,7 @@ class InodePages(plugins.PluginInterface):
_required_framework_version = (2, 0, 0)
_version = (2, 0, 1)
_version = (2, 0, 2)
@classmethod
def get_requirements(cls) -> List[interfaces.configuration.RequirementInterface]:
@@ -436,8 +436,9 @@ class InodePages(plugins.PluginInterface):
),
]
@staticmethod
@classmethod
def write_inode_content_to_file(
cls,
inode: interfaces.objects.ObjectInterface,
filename: str,
open_method: Type[interfaces.plugins.FileHandlerInterface],
@@ -18,7 +18,7 @@ class VmaYaraScan(interfaces.plugins.PluginInterface):
"""Scans all virtual memory areas for tasks using yara."""
_required_framework_version = (2, 4, 0)
_version = (1, 0, 2)
_version = (1, 0, 3)
@classmethod
def get_requirements(cls) -> List[interfaces.configuration.RequirementInterface]:
@@ -105,8 +105,9 @@ class VmaYaraScan(interfaces.plugins.PluginInterface):
value,
)
@staticmethod
@classmethod
def get_vma_maps(
cls,
task: interfaces.objects.ObjectInterface,
) -> Iterable[Tuple[int, int]]:
"""Creates a map of start/end addresses for each virtual memory area in a task.
@@ -22,7 +22,7 @@ class Cachedump(interfaces.plugins.PluginInterface):
"""Dumps lsa secrets from memory"""
_required_framework_version = (2, 0, 0)
_version = (1, 0, 0)
_version = (1, 0, 1)
@classmethod
def get_requirements(cls):
@@ -43,16 +43,16 @@ class Cachedump(interfaces.plugins.PluginInterface):
),
]
@staticmethod
@classmethod
def get_nlkm(
sechive: registry.RegistryHive, lsakey: bytes, is_vista_or_later: bool
cls, sechive: registry.RegistryHive, lsakey: bytes, is_vista_or_later: bool
):
return lsadump.Lsadump.get_secret_by_name(
sechive, "NL$KM", lsakey, is_vista_or_later
)
@staticmethod
def decrypt_hash(edata: bytes, nlkm: bytes, ch, xp: bool):
@classmethod
def decrypt_hash(cls, edata: bytes, nlkm: bytes, ch, xp: bool):
if xp:
hmac_md5 = HMAC.new(nlkm, ch)
rc4key = hmac_md5.digest()
@@ -69,8 +69,8 @@ class Cachedump(interfaces.plugins.PluginInterface):
data += aes.decrypt(buf)
return data
@staticmethod
def parse_cache_entry(cache_data: bytes) -> Tuple[int, int, int, bytes, bytes]:
@classmethod
def parse_cache_entry(cls, cache_data: bytes) -> Tuple[int, int, int, bytes, bytes]:
(uname_len, domain_len) = unpack("<HH", cache_data[:4])
if len(cache_data[60:62]) == 0:
return (uname_len, domain_len, 0, b"", b"")
@@ -79,9 +79,9 @@ class Cachedump(interfaces.plugins.PluginInterface):
enc_data = cache_data[96:]
return (uname_len, domain_len, domain_name_len, enc_data, ch)
@staticmethod
@classmethod
def parse_decrypted_cache(
dec_data: bytes, uname_len: int, domain_len: int, domain_name_len: int
cls, dec_data: bytes, uname_len: int, domain_len: int, domain_name_len: int
) -> Tuple[str, str, str, bytes]:
"""Get the data from the cache and separate it into the username, domain name, and hash data"""
uname_offset = 72
@@ -53,7 +53,7 @@ class DirectSystemCalls(interfaces.plugins.PluginInterface):
"""Detects the Direct System Call technique used to bypass EDRs"""
_required_framework_version = (2, 4, 0)
_version = (1, 0, 0)
_version = (1, 0, 1)
# DLLs that are expected to host system call invocations
valid_syscall_handlers = ("ntdll.dll", "win32u.dll")
@@ -200,8 +200,8 @@ class DirectSystemCalls(interfaces.plugins.PluginInterface):
return disasm_bytes, end_inst
@staticmethod
def get_disasm_function(architecture: str) -> Callable:
@classmethod
def get_disasm_function(cls, architecture: str) -> Callable:
"""
Returns the disassembly handler for the given architecture
.detail is used to get full instruction information
@@ -284,8 +284,9 @@ class DirectSystemCalls(interfaces.plugins.PluginInterface):
return None
@staticmethod
@classmethod
def get_vad_maps(
cls,
task: interfaces.objects.ObjectInterface,
) -> List[Tuple[int, int, str]]:
"""Creates a map of start/end addresses within a virtual address
@@ -310,9 +311,9 @@ class DirectSystemCalls(interfaces.plugins.PluginInterface):
return vads
@staticmethod
@classmethod
def get_range_path(
ranges: List[Tuple[int, int, str]], address: int
cls, ranges: List[Tuple[int, int, str]], address: int
) -> Optional[str]:
"""
Returns the path for the range holding `address`, if found
@@ -22,7 +22,7 @@ class MFTScan(interfaces.plugins.PluginInterface, timeliner.TimeLinerInterface):
_required_framework_version = (2, 0, 0)
_version = (2, 0, 0)
_version = (2, 0, 1)
@classmethod
def get_requirements(cls):
@@ -37,8 +37,9 @@ class MFTScan(interfaces.plugins.PluginInterface, timeliner.TimeLinerInterface):
),
]
@staticmethod
@classmethod
def enumerate_mft_records(
cls,
context: interfaces.context.ContextInterface,
config_path: str,
primary_layer_name: str,
@@ -128,8 +129,9 @@ class MFTScan(interfaces.plugins.PluginInterface, timeliner.TimeLinerInterface):
layer_name=layer.name,
)
@staticmethod
@classmethod
def parse_mft_records(
cls,
record_map: Dict[int, Tuple[str, int, int]],
mft_record: interfaces.objects.ObjectInterface,
attr: interfaces.objects.ObjectInterface,
@@ -191,8 +193,9 @@ class MFTScan(interfaces.plugins.PluginInterface, timeliner.TimeLinerInterface):
file_name,
)
@staticmethod
@classmethod
def parse_data_record(
cls,
mft_record: interfaces.objects.ObjectInterface,
attr: interfaces.objects.ObjectInterface,
record_map: Dict[int, Tuple[str, int, int]],
@@ -325,7 +328,7 @@ class ADS(interfaces.plugins.PluginInterface):
_required_framework_version = (2, 7, 0)
_version = (1, 0, 0)
_version = (1, 0, 1)
@classmethod
def get_requirements(cls):
@@ -343,8 +346,9 @@ class ADS(interfaces.plugins.PluginInterface):
),
]
@staticmethod
@classmethod
def parse_ads_data_records(
cls,
record_map: Dict[int, Tuple[str, int, int]],
mft_record: interfaces.objects.ObjectInterface,
attr: interfaces.objects.ObjectInterface,
@@ -394,7 +398,7 @@ class ResidentData(interfaces.plugins.PluginInterface):
_required_framework_version = (2, 7, 0)
_version = (1, 0, 0)
_version = (1, 0, 1)
@classmethod
def get_requirements(cls):
@@ -412,8 +416,9 @@ class ResidentData(interfaces.plugins.PluginInterface):
),
]
@staticmethod
@classmethod
def parse_first_data_records(
cls,
record_map: Dict[int, Tuple[str, int, int]],
mft_record: interfaces.objects.ObjectInterface,
attr: interfaces.objects.ObjectInterface,
@@ -23,7 +23,7 @@ class NetScan(interfaces.plugins.PluginInterface, timeliner.TimeLinerInterface):
"""Scans for network objects present in a particular windows memory image."""
_required_framework_version = (2, 0, 0)
_version = (1, 0, 0)
_version = (1, 0, 1)
@classmethod
def get_requirements(cls):
@@ -50,9 +50,9 @@ class NetScan(interfaces.plugins.PluginInterface, timeliner.TimeLinerInterface):
),
]
@staticmethod
@classmethod
def create_netscan_constraints(
context: interfaces.context.ContextInterface, symbol_table: str
cls, context: interfaces.context.ContextInterface, symbol_table: str
) -> List[poolscanner.PoolConstraint]:
"""Creates a list of Pool Tag Constraints for network objects.
@@ -331,9 +331,9 @@ class PESymbols(interfaces.plugins.PluginInterface):
return pe_ret
@staticmethod
@classmethod
def range_info_for_address(
ranges: ranges_type, address: int
cls, ranges: ranges_type, address: int
) -> Optional[range_type]:
"""
Helper for getting the range information for an address.
@@ -352,8 +352,8 @@ class PESymbols(interfaces.plugins.PluginInterface):
return None
@staticmethod
def filepath_for_address(ranges: ranges_type, address: int) -> Optional[str]:
@classmethod
def filepath_for_address(cls, ranges: ranges_type, address: int) -> Optional[str]:
"""
Helper to get the file path for an address
@@ -370,8 +370,8 @@ class PESymbols(interfaces.plugins.PluginInterface):
return None
@staticmethod
def filename_for_path(filepath: str) -> str:
@classmethod
def filename_for_path(cls, filepath: str) -> str:
"""
Consistent way to get the filename regardless of platform
@@ -383,8 +383,9 @@ class PESymbols(interfaces.plugins.PluginInterface):
"""
return ntpath.basename(filepath).lower()
@staticmethod
@classmethod
def addresses_for_process_symbols(
cls,
context: interfaces.context.ContextInterface,
config_path: str,
layer_name: str,
@@ -417,8 +418,9 @@ class PESymbols(interfaces.plugins.PluginInterface):
return found_symbols
@staticmethod
@classmethod
def path_and_symbol_for_address(
cls,
context: interfaces.context.ContextInterface,
config_path: str,
collected_modules: collected_modules_type,
@@ -734,8 +736,9 @@ class PESymbols(interfaces.plugins.PluginInterface):
return found, remaining
@staticmethod
@classmethod
def find_symbols(
cls,
context: interfaces.context.ContextInterface,
config_path: str,
wanted_modules: PESymbolFinder.cached_value_dict,
@@ -776,8 +779,9 @@ class PESymbols(interfaces.plugins.PluginInterface):
return found_symbols, missing_symbols
@staticmethod
@classmethod
def get_kernel_modules(
cls,
context: interfaces.context.ContextInterface,
layer_name: str,
symbol_table: str,
@@ -838,8 +842,9 @@ class PESymbols(interfaces.plugins.PluginInterface):
return found_modules
@staticmethod
@classmethod
def get_vads_for_process_cache(
cls,
vads_cache: Dict[int, ranges_type],
owner_proc: interfaces.objects.ObjectInterface,
) -> Optional[ranges_type]:
@@ -866,8 +871,9 @@ class PESymbols(interfaces.plugins.PluginInterface):
return vads
@staticmethod
@classmethod
def get_proc_vads_with_file_paths(
cls,
proc: interfaces.objects.ObjectInterface,
) -> ranges_type:
"""
@@ -929,8 +935,9 @@ class PESymbols(interfaces.plugins.PluginInterface):
yield proc, proc_layer_name, vads
@staticmethod
@classmethod
def get_process_modules(
cls,
context: interfaces.context.ContextInterface,
layer_name: str,
symbol_table: str,
@@ -127,8 +127,8 @@ class PoolHeaderScanner(interfaces.layers.ScannerInterface):
class PoolScanner(plugins.PluginInterface):
"""A generic pool scanner plugin."""
_version = (1, 0, 0)
_required_framework_version = (2, 0, 0)
_version = (1, 0, 1)
@classmethod
def get_requirements(cls) -> List[interfaces.configuration.RequirementInterface]:
@@ -181,9 +181,9 @@ class PoolScanner(plugins.PluginInterface):
),
)
@staticmethod
@classmethod
def builtin_constraints(
symbol_table: str, tags_filter: Optional[List[bytes]] = None
cls, symbol_table: str, tags_filter: Optional[List[bytes]] = None
) -> List[PoolConstraint]:
"""Get built-in PoolConstraints given a list of pool tags.
@@ -24,6 +24,7 @@ class ShimcacheMem(interfaces.plugins.PluginInterface, timeliner.TimeLinerInterf
"""Reads Shimcache entries from the ahcache.sys AVL tree"""
_required_framework_version = (2, 0, 0)
_version = (1, 0, 1)
# These checks must be completed from newest -> oldest OS version.
_win_version_file_map: List[Tuple[versions.OsDistinguisher, bool, str]] = [
@@ -74,8 +75,9 @@ class ShimcacheMem(interfaces.plugins.PluginInterface, timeliner.TimeLinerInterf
),
]
@staticmethod
@classmethod
def create_shimcache_table(
cls,
context: interfaces.context.ContextInterface,
symbol_table: str,
config_path: str,
@@ -35,7 +35,7 @@ class SvcScan(interfaces.plugins.PluginInterface):
"""Scans for windows services."""
_required_framework_version = (2, 0, 0)
_version = (3, 0, 1)
_version = (3, 0, 2)
def __init__(self, *args, **kwargs):
super().__init__(*args, **kwargs)
@@ -61,8 +61,9 @@ class SvcScan(interfaces.plugins.PluginInterface):
),
]
@staticmethod
@classmethod
def get_record_tuple(
cls,
service_record: interfaces.objects.ObjectInterface,
binary_info: ServiceBinaryInfo,
):
@@ -22,7 +22,7 @@ class UnloadedModules(interfaces.plugins.PluginInterface, timeliner.TimeLinerInt
"""Lists the unloaded kernel modules."""
_required_framework_version = (2, 0, 0)
_version = (1, 0, 0)
_version = (1, 0, 1)
@classmethod
def get_requirements(cls) -> List[interfaces.configuration.RequirementInterface]:
@@ -34,8 +34,9 @@ class UnloadedModules(interfaces.plugins.PluginInterface, timeliner.TimeLinerInt
),
]
@staticmethod
@classmethod
def create_unloadedmodules_table(
cls,
context: interfaces.context.ContextInterface,
symbol_table: str,
config_path: str,
@@ -18,7 +18,7 @@ class VadYaraScan(interfaces.plugins.PluginInterface):
"""Scans all the Virtual Address Descriptor memory maps using yara."""
_required_framework_version = (2, 4, 0)
_version = (1, 1, 1)
_version = (1, 1, 2)
@classmethod
def get_requirements(cls) -> List[interfaces.configuration.RequirementInterface]:
@@ -84,7 +84,7 @@ class VadYaraScan(interfaces.plugins.PluginInterface):
if not vad_maps_to_scan:
vollog.warning(
f"No VADs were found for task {task.UniqueProcessID}, not scanning"
f"No VADs were found for task {task.UniqueProcessId}, not scanning"
)
continue
@@ -104,8 +104,9 @@ class VadYaraScan(interfaces.plugins.PluginInterface):
value,
)
@staticmethod
@classmethod
def get_vad_maps(
cls,
task: interfaces.objects.ObjectInterface,
) -> Iterable[Tuple[int, int]]:
"""Creates a map of start/end addresses within a virtual address
+7 -7
View File
@@ -37,7 +37,7 @@ except ImportError:
class YaraScanner(interfaces.layers.ScannerInterface):
_version = (2, 1, 0)
_version = (2, 1, 1)
# yara.Rules isn't exposed, so we can't type this properly
def __init__(self, rules) -> None:
@@ -79,23 +79,23 @@ class YaraScanner(interfaces.layers.ScannerInterface):
for offset, name, value in match.strings:
yield (offset + data_offset, match.rule, name, value)
@staticmethod
def get_rule(rule):
@classmethod
def get_rule(cls, rule):
if USE_YARA_X:
return yara_x.compile(f"rule r1 {{strings: $a = {rule} condition: $a}}")
return yara.compile(
sources={"n": f"rule r1 {{strings: $a = {rule} condition: $a}}"}
)
@staticmethod
def from_compiled_file(filepath):
@classmethod
def from_compiled_file(cls, filepath):
with resources.ResourceAccessor().open(filepath, "rb") as fp:
if USE_YARA_X:
return yara_x.Rules.deserialize_from(file=fp)
return yara.load(file=fp)
@staticmethod
def from_file(filepath):
@classmethod
def from_file(cls, filepath):
with resources.ResourceAccessor().open(filepath, "rb") as fp:
if USE_YARA_X:
return yara_x.compile(fp.read().decode())
@@ -88,7 +88,7 @@ class LinuxKernelIntermedSymbols(intermed.IntermediateSymbolTable):
class LinuxUtilities(interfaces.configuration.VersionableInterface):
"""Class with multiple useful linux functions."""
_version = (2, 2, 1)
_version = (2, 3, 0)
_required_framework_version = (2, 0, 0)
framework.require_interface_version(*_required_framework_version)
@@ -118,8 +118,8 @@ class LinuxUtilities(interfaces.configuration.VersionableInterface):
Args:
task (task_struct): A reference task
mnt (vfsmount or mount): A mounted filesystem or a mount point.
- kernels < 3.3.8 type is 'vfsmount'
- kernels >= 3.3.8 type is 'mount'
- kernels < 3.3 type is 'vfsmount'
- kernels >= 3.3 type is 'mount'
Returns:
str: Pathname of the mount point relative to the task's root directory.
@@ -141,14 +141,28 @@ class LinuxUtilities(interfaces.configuration.VersionableInterface):
rdentry (dentry *): A pointer to the root dentry
rmnt (vfsmount *): A pointer to the root vfsmount
dentry (dentry *): A pointer to the dentry
vfsmnt (vfsmount *): A pointer to the vfsmount
vfsmnt (vfsmount/vfsmount *): A vfsmount object (kernels >= 3.3) or a
vfsmount pointer (kernels < 3.3)
Returns:
str: Pathname of the mount point or file
"""
if not (rdentry and rdentry.is_readable() and rmnt and rmnt.is_readable()):
return ""
if isinstance(vfsmnt, objects.Pointer) and not (
vfsmnt and vfsmnt.is_readable()
):
# vfsmnt can be the vfsmount object itself (>=3.3) or a vfsmount * (<3.3)
return ""
path_reversed = []
while dentry != rdentry or not vfsmnt.is_equal(rmnt):
while (
dentry
and dentry.is_readable()
and (dentry != rdentry or not vfsmnt.is_equal(rmnt))
):
if dentry == vfsmnt.get_mnt_root() or dentry.is_root():
# Escaped?
if dentry != vfsmnt.get_mnt_root():
@@ -447,6 +461,10 @@ class LinuxUtilities(interfaces.configuration.VersionableInterface):
type_dec = vmlinux.get_type(type_name)
member_offset = type_dec.relative_child_offset(member_name)
container_addr = addr - member_offset
layer = vmlinux.context.layers[vmlinux.layer_name]
if not layer.is_valid(container_addr):
return None
return vmlinux.object(
object_type=type_name, offset=container_addr, absolute=True
)
@@ -1211,19 +1211,15 @@ class struct_file(objects.StructType):
"""Returns a pointer to the dentry associated with this file"""
if self.has_member("f_path"):
return self.f_path.dentry
elif self.has_member("f_dentry"):
return self.f_dentry
else:
raise AttributeError("Unable to find file -> dentry")
raise AttributeError("Unable to find file -> dentry")
def get_vfsmnt(self) -> interfaces.objects.ObjectInterface:
"""Returns the fs (vfsmount) where this file is mounted"""
if self.has_member("f_path"):
return self.f_path.mnt
elif self.has_member("f_vfsmnt"):
return self.f_vfsmnt
else:
raise AttributeError("Unable to find file -> vfs mount")
raise AttributeError("Unable to find file -> vfs mount")
def get_inode(self) -> interfaces.objects.ObjectInterface:
"""Returns an inode associated with this file"""
@@ -1461,9 +1457,9 @@ class mount(objects.StructType):
A dentry pointer
"""
vfsmnt = self.get_vfsmnt_current()
dentry = vfsmnt.mnt_root
dentry_pointer = vfsmnt.mnt_root
return dentry
return dentry_pointer
def get_dentry_parent(self):
"""Returns the parent root of the mounted tree
@@ -1571,39 +1567,38 @@ class vfsmount(objects.StructType):
)
def _is_kernel_prior_to_struct_mount(self) -> bool:
"""Helper to distinguish between kernels prior to version 3.3.8 that
lacked the 'mount' structure and later versions that have it.
"""Helper to distinguish between kernels prior to version 3.3 which lacked the
'mount' struct, versus later versions that include it.
See 7d6fec45a5131918b51dcd76da52f2ec86a85be6.
The 'mnt_parent' member was moved from struct 'vfsmount' to struct
'mount' when the latter was introduced.
Alternatively, vmlinux.has_type('mount') can be used here but it is faster.
# Following that commit, also in kernel version 3.3 (3376f34fff5be9954fd9a9c4fd68f4a0a36d480e),
# the 'mnt_parent' member was relocated from the 'vfsmount' struct to the newly
# introduced 'mount' struct.
Returns:
bool: 'True' if the kernel
'True' if the kernel lacks the 'mount' struct, typically indicating kernel < 3.3.
"""
return self.has_member("mnt_parent")
return not self._context.symbol_space.has_type("mount")
def is_equal(self, vfsmount_ptr) -> bool:
"""Helper to make sure it is comparing two pointers to 'vfsmount'.
Depending on the kernel version, the calling object (self) could be
a 'vfsmount \\*' (<3.3.8) or a 'vfsmount' (>=3.3.8). This way we trust
in the framework "auto" dereferencing ability to assure that when we
reach this point 'self' will be a 'vfsmount' already and self.vol.offset
Depending on the kernel version, see 3376f34fff5be9954fd9a9c4fd68f4a0a36d480e,
the calling object (self) could be a 'vfsmount \\*' (<3.3) or a 'vfsmount' (>=3.3).
This way we trust in the framework "auto" dereferencing ability to assure that
when we reach this point 'self' will be a 'vfsmount' already and self.vol.offset
a 'vfsmount \\*' and not a 'vfsmount \\*\\*'. The argument must be a 'vfsmount \\*'.
Typically, it's called from do_get_path().
Args:
vfsmount_ptr (vfsmount *): A pointer to a 'vfsmount'
vfsmount_ptr: A pointer to a 'vfsmount'
Raises:
exceptions.VolatilityException: If vfsmount_ptr is not a 'vfsmount \\*'
Returns:
bool: 'True' if the given argument points to the the same 'vfsmount'
as 'self'.
'True' if the given argument points to the same 'vfsmount' as 'self'.
"""
if isinstance(vfsmount_ptr, objects.Pointer):
return self.vol.offset == vfsmount_ptr
@@ -1612,13 +1607,14 @@ class vfsmount(objects.StructType):
"Unexpected argument type. It has to be a 'vfsmount *'"
)
def _get_real_mnt(self):
def _get_real_mnt(self) -> interfaces.objects.ObjectInterface:
"""Gets the struct 'mount' containing this 'vfsmount'.
It should be only called from kernels >= 3.3.8 when 'struct mount' was introduced.
It should be only called from kernels >= 3.3 when 'struct mount' was introduced.
See 7d6fec45a5131918b51dcd76da52f2ec86a85be6
Returns:
mount: the struct 'mount' containing this 'vfsmount'.
The 'mount' object containing this 'vfsmount'.
"""
vmlinux = linux.LinuxUtilities.get_module_from_volobj_type(self._context, self)
return linux.LinuxUtilities.container_of(
@@ -1637,8 +1633,8 @@ class vfsmount(objects.StructType):
"""Gets the parent fs (vfsmount) to where it's mounted on
Returns:
For kernels < 3.3.8: A vfsmount pointer
For kernels >= 3.3.8: A vfsmount object
For kernels < 3.3: A vfsmount pointer
For kernels >= 3.3: A vfsmount object
"""
if self._is_kernel_prior_to_struct_mount():
return self.get_mnt_parent()
@@ -1671,8 +1667,8 @@ class vfsmount(objects.StructType):
"""Gets the mnt_parent member.
Returns:
For kernels < 3.3.8: A vfsmount pointer
For kernels >= 3.3.8: A mount pointer
For kernels < 3.3: A vfsmount pointer
For kernels >= 3.3: A mount pointer
"""
if self._is_kernel_prior_to_struct_mount():
return self.mnt_parent
@@ -1743,8 +1739,10 @@ class kobject(objects.StructType):
class mnt_namespace(objects.StructType):
def get_inode(self):
if self.has_member("proc_inum"):
# 98f842e675f96ffac96e6c50315790912b2812be 3.8 <= kernels < 3.19
return self.proc_inum
elif self.has_member("ns") and self.ns.has_member("inum"):
# kernels >= 3.19 435d5f4bb2ccba3b791d9ef61d2590e30b8e806e
return self.ns.inum
else:
raise AttributeError("Unable to find mnt_namespace inode")