mirror of
https://github.com/volatilityfoundation/volatility3.git
synced 2026-09-12 12:47:39 +02:00
Make several small typing information fixes.
This commit is contained in:
@@ -8,8 +8,9 @@ import random
|
||||
import string
|
||||
import sys
|
||||
from functools import wraps
|
||||
from typing import Any, List, Tuple
|
||||
from typing import Any, List, Tuple, Dict
|
||||
|
||||
from volatility.framework.interfaces.renderers import Column
|
||||
from volatility.framework.renderers import format_hints
|
||||
|
||||
vollog = logging.getLogger(__name__)
|
||||
@@ -235,7 +236,7 @@ class PrettyTextRenderer(CLIRenderer):
|
||||
tree_indent_column = ''.join(random.choice(string.ascii_uppercase + string.digits) for _ in range(20))
|
||||
max_column_widths = dict([(column.name, len(column.name)) for column in grid.columns])
|
||||
|
||||
def visitor(node, accumulator):
|
||||
def visitor(node, accumulator: List[Tuple[int, Dict[Column, bytes]]]) -> List[Tuple[int, Dict[Column, bytes]]]:
|
||||
# Nodes always have a path value, giving them a path_depth of at least 1, we use max just in case
|
||||
max_column_widths[tree_indent_column] = max(max_column_widths.get(tree_indent_column, 0), node.path_depth)
|
||||
line = {}
|
||||
@@ -249,7 +250,7 @@ class PrettyTextRenderer(CLIRenderer):
|
||||
accumulator.append((node.path_depth, line))
|
||||
return accumulator
|
||||
|
||||
final_output = [] # type: List[Tuple[int, List]]
|
||||
final_output = [] # type: List[Tuple[int, Dict[Column, bytes]]]
|
||||
grid.populate(visitor, final_output)
|
||||
|
||||
# Always align the tree to the left
|
||||
|
||||
@@ -45,7 +45,7 @@ class Psaux(plugins.PluginInterface):
|
||||
|
||||
task_name = utility.array_to_string(task.p_comm)
|
||||
|
||||
args = []
|
||||
args = [] # type: List[bytes]
|
||||
|
||||
while argc > 0:
|
||||
try:
|
||||
|
||||
@@ -3,7 +3,7 @@
|
||||
#
|
||||
|
||||
import logging
|
||||
from typing import Callable, Iterable, List
|
||||
from typing import Callable, Iterable, List, Dict
|
||||
|
||||
from volatility.framework import renderers, interfaces, contexts
|
||||
from volatility.framework.automagic import mac
|
||||
@@ -77,7 +77,7 @@ class PsList(interfaces.plugins.PluginInterface):
|
||||
|
||||
proc = kernel.object_from_symbol(symbol_name = "allproc").lh_first
|
||||
|
||||
seen = {}
|
||||
seen = {} # type: Dict[int, int]
|
||||
while proc is not None and proc.vol.offset != 0:
|
||||
if proc.vol.offset in seen:
|
||||
vollog.log(logging.INFO, "Recursive process list detected (a result of non-atomic acquisition).")
|
||||
|
||||
@@ -3,7 +3,7 @@
|
||||
#
|
||||
|
||||
import logging
|
||||
from typing import Callable, Iterable
|
||||
from typing import Callable, Dict, Iterable
|
||||
|
||||
from volatility.framework import interfaces, contexts
|
||||
from volatility.framework.automagic import mac
|
||||
@@ -40,7 +40,7 @@ class Tasks(pslist.PsList):
|
||||
|
||||
queue_entry = kernel.object_from_symbol(symbol_name = "tasks")
|
||||
|
||||
seen = {}
|
||||
seen = {} # type: Dict[int, int]
|
||||
for task in queue_entry.walk_list(queue_entry, "tasks", "task"):
|
||||
if task.vol.offset in seen:
|
||||
vollog.log(logging.INFO, "Recursive process list detected (a result of non-atomic acquisition).")
|
||||
|
||||
@@ -3,7 +3,7 @@
|
||||
#
|
||||
|
||||
import logging
|
||||
from typing import List, Iterable, Tuple
|
||||
from typing import List, Iterable, Tuple, Optional, Union
|
||||
|
||||
import volatility.framework.interfaces.plugins as interfaces_plugins
|
||||
from volatility.framework import constants, exceptions, renderers, interfaces, symbols
|
||||
@@ -62,7 +62,7 @@ class Callbacks(interfaces_plugins.PluginInterface):
|
||||
|
||||
@classmethod
|
||||
def list_notify_routines(cls, context: interfaces.context.ContextInterface, layer_name: str, symbol_table: str,
|
||||
callback_table_name: str) -> Iterable[Tuple[str, int, str]]:
|
||||
callback_table_name: str) -> Iterable[Tuple[str, int, Optional[str]]]:
|
||||
"""Lists all kernel notification routines.
|
||||
|
||||
Args:
|
||||
@@ -114,7 +114,7 @@ class Callbacks(interfaces_plugins.PluginInterface):
|
||||
|
||||
@classmethod
|
||||
def list_registry_callbacks(cls, context: interfaces.context.ContextInterface, layer_name: str, symbol_table: str,
|
||||
callback_table_name: str) -> Iterable[Tuple[str, int, str]]:
|
||||
callback_table_name: str) -> Iterable[Tuple[str, int, None]]:
|
||||
"""Lists all registry callbacks.
|
||||
|
||||
Args:
|
||||
@@ -193,7 +193,8 @@ class Callbacks(interfaces_plugins.PluginInterface):
|
||||
|
||||
try:
|
||||
component = ntkrnlmp.object(
|
||||
"string", absolute = True, offset = callback.Component, max_length = 64, errors = "replace")
|
||||
"string", absolute = True, offset = callback.Component, max_length = 64, errors = "replace"
|
||||
) # type: Union[interfaces.renderers.BaseAbsentValue, interfaces.objects.ObjectInterface]
|
||||
except exceptions.InvalidAddressException:
|
||||
component = renderers.UnreadableValue()
|
||||
|
||||
|
||||
@@ -359,8 +359,8 @@ class PoolScanner(plugins.PluginInterface):
|
||||
|
||||
cookie = handles.Handles.find_cookie(context = context, layer_name = layer_name, symbol_table = symbol_table)
|
||||
|
||||
is_windows_10 = cls.is_windows_10(context = context, symbol_table = symbol_table)
|
||||
is_windows_8_or_later = cls.is_windows_8_or_later(context = context, symbol_table = symbol_table)
|
||||
is_windows_10 = cls.is_windows_10(context, symbol_table)
|
||||
is_windows_8_or_later = cls.is_windows_8_or_later(context, symbol_table)
|
||||
|
||||
# start off with the primary virtual layer
|
||||
scan_layer = layer_name
|
||||
@@ -434,7 +434,7 @@ class PoolScanner(plugins.PluginInterface):
|
||||
# We have to manually load a symbol table
|
||||
|
||||
if symbols.symbol_table_is_64bit(context, symbol_table):
|
||||
is_win_7 = cls.is_windows_7(context = context, symbol_table = symbol_table)
|
||||
is_win_7 = cls.is_windows_7(context, symbol_table)
|
||||
if is_win_7:
|
||||
pool_header_json_filename = "poolheader-x64-win7"
|
||||
else:
|
||||
|
||||
@@ -4,7 +4,7 @@
|
||||
|
||||
import datetime
|
||||
import logging
|
||||
from typing import List, Sequence, Iterable, Tuple
|
||||
from typing import List, Sequence, Iterable, Tuple, Union
|
||||
|
||||
from volatility.framework import objects, renderers, exceptions, interfaces, constants
|
||||
from volatility.framework.configuration import requirements
|
||||
@@ -36,8 +36,8 @@ class PrintKey(interfaces.plugins.PluginInterface):
|
||||
]
|
||||
|
||||
@classmethod
|
||||
def key_iterator(cls, hive: RegistryHive, node_path: Sequence[objects.StructType] = None,
|
||||
recurse: bool = False) -> Iterable[Tuple[int, bool, datetime.datetime, str, bool, bytes]]:
|
||||
def key_iterator(cls, hive: RegistryHive, node_path: Sequence[objects.StructType] = None, recurse: bool = False
|
||||
) -> Iterable[Tuple[int, bool, datetime.datetime, str, bool, interfaces.objects.ObjectInterface]]:
|
||||
"""Walks through a set of nodes from a given node (last one in
|
||||
node_path). Avoids loops by not traversing into nodes already present
|
||||
in the node_path.
|
||||
@@ -112,7 +112,7 @@ class PrintKey(interfaces.plugins.PluginInterface):
|
||||
value_node_name = renderers.UnreadableValue()
|
||||
|
||||
try:
|
||||
value_data = str(node.decode_data())
|
||||
value_data = str(node.decode_data()) # type: Union[interfaces.renderers.BaseAbsentValue, str]
|
||||
except (ValueError, exceptions.InvalidAddressException, RegistryFormatException) as excp:
|
||||
vollog.debug(excp)
|
||||
value_data = renderers.UnreadableValue()
|
||||
|
||||
@@ -11,7 +11,7 @@ import os
|
||||
import pathlib
|
||||
import zipfile
|
||||
from abc import ABCMeta
|
||||
from typing import Any, Dict, Generator, Iterable, List, Optional, Type, Tuple
|
||||
from typing import Any, Dict, Generator, Iterable, List, Optional, Type, Tuple, Mapping
|
||||
|
||||
import volatility
|
||||
import volatility.framework.layers.resources
|
||||
@@ -208,7 +208,7 @@ class IntermediateSymbolTable(interfaces.symbols.SymbolTableInterface):
|
||||
filename: str,
|
||||
native_types: Optional[interfaces.symbols.NativeTableInterface] = None,
|
||||
table_mapping: Optional[Dict[str, str]] = None,
|
||||
class_types: Optional[Dict[str, Type[interfaces.objects.ObjectInterface]]] = None) -> str:
|
||||
class_types: Optional[Mapping[str, Type[interfaces.objects.ObjectInterface]]] = None) -> str:
|
||||
"""Takes a context and loads an intermediate symbol table based on a
|
||||
filename.
|
||||
|
||||
|
||||
@@ -70,6 +70,7 @@ class _HMAP_ENTRY(objects.StructType):
|
||||
except AttributeError:
|
||||
return self.BlockAddress
|
||||
|
||||
|
||||
class _CMHIVE(objects.StructType):
|
||||
|
||||
def get_name(self) -> Optional[interfaces.objects.ObjectInterface]:
|
||||
@@ -205,7 +206,7 @@ class _CM_KEY_NODE(objects.StructType):
|
||||
"""Since this is just a casting convenience, it can be a property."""
|
||||
return self.Name.cast("string", max_length = self.NameLength, encoding = "latin-1")
|
||||
|
||||
def get_key_path(self) -> interfaces.objects.ObjectInterface:
|
||||
def get_key_path(self) -> str:
|
||||
reg = self._context.layers[self.vol.layer_name]
|
||||
if not isinstance(reg, RegistryHive):
|
||||
raise TypeError("Key was not instantiated on a RegistryHive layer")
|
||||
@@ -277,10 +278,10 @@ class _CM_KEY_VALUE(objects.StructType):
|
||||
return output
|
||||
if self_type == RegValueTypes.REG_MULTI_SZ:
|
||||
return str(data, encoding = "utf-16-le").split("\x00")[0]
|
||||
if self_type in [RegValueTypes.REG_BINARY,
|
||||
RegValueTypes.REG_FULL_RESOURCE_DESCRIPTOR,
|
||||
RegValueTypes.REG_RESOURCE_LIST,
|
||||
RegValueTypes.REG_RESOURCE_REQUIREMENTS_LIST]:
|
||||
if self_type in [
|
||||
RegValueTypes.REG_BINARY, RegValueTypes.REG_FULL_RESOURCE_DESCRIPTOR, RegValueTypes.REG_RESOURCE_LIST,
|
||||
RegValueTypes.REG_RESOURCE_REQUIREMENTS_LIST
|
||||
]:
|
||||
return data
|
||||
if self_type == RegValueTypes.REG_NONE:
|
||||
return ''
|
||||
|
||||
Reference in New Issue
Block a user