Shift exposed staticmethods to classmethods

This commit is contained in:
Mike Auty
2025-01-26 13:44:33 +00:00
parent 69fc7fe723
commit 1ac2dbc49c
16 changed files with 93 additions and 70 deletions
@@ -35,7 +35,9 @@ class CapabilitiesData:
cap_permitted: interfaces.objects.ObjectInterface
cap_effective: interfaces.objects.ObjectInterface
cap_bset: interfaces.objects.ObjectInterface
cap_ambient: interfaces.objects.ObjectInterface
cap_ambient: (
interfaces.objects.ObjectInterface | interfaces.renderers.BaseAbsentValue
)
def astuple(self) -> Tuple:
"""Returns a shallow copy of the capability sets in a tuple.
@@ -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]:
@@ -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.
@@ -244,7 +244,7 @@ class PESymbols(interfaces.plugins.PluginInterface):
_required_framework_version = (2, 7, 0)
_version = (1, 0, 0)
_version = (1, 0, 1)
# used for special handling of the kernel PDB file. See later notes
os_module_name = "ntoskrnl.exe"
@@ -330,9 +330,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.
@@ -351,8 +351,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
@@ -369,8 +369,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
@@ -382,8 +382,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,
@@ -416,8 +417,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,
@@ -733,8 +735,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,
@@ -775,8 +778,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,
@@ -837,8 +841,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]:
@@ -865,8 +870,9 @@ class PESymbols(interfaces.plugins.PluginInterface):
return vads
@staticmethod
@classmethod
def get_proc_vads_with_file_paths(
cls,
proc: interfaces.objects.ObjectInterface,
) -> ranges_type:
"""
@@ -928,8 +934,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]:
@@ -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())