Minor improvements for mypy

This commit is contained in:
TheMythologist
2024-12-18 15:18:33 +08:00
parent 197d24567a
commit d7f678879d
43 changed files with 127 additions and 94 deletions
+2 -1
View File
@@ -32,6 +32,7 @@ dev = [
"jsonschema>=4.23.0,<5",
"pyinstaller>=6.11.0,<7",
"pyinstaller-hooks-contrib>=2024.9",
"types-jsonschema>=4.23.0,<5",
]
test = [
@@ -68,7 +69,7 @@ include = ["volatility3*"]
mypy_path = "./stubs"
show_traceback = true
[tool.mypy.overrides]
[[tool.mypy.overrides]]
ignore_missing_imports = true
[tool.ruff]
+3 -3
View File
@@ -19,7 +19,7 @@ import os
import sys
import tempfile
import traceback
from typing import Any, Dict, List, Tuple, Type, Union
from typing import Any, Dict, List, Optional, Tuple, Type, Union
from urllib import parse, request
try:
@@ -64,7 +64,7 @@ class PrintedProgress:
def __init__(self):
self._max_message_len = 0
def __call__(self, progress: Union[int, float], description: str = None):
def __call__(self, progress: Union[int, float], description: Optional[str] = None):
"""A simple function for providing text-based feedback.
.. warning:: Only for development use.
@@ -81,7 +81,7 @@ class PrintedProgress:
class MuteProgress(PrintedProgress):
"""A dummy progress handler that produces no output when called."""
def __call__(self, progress: Union[int, float], description: str = None):
def __call__(self, progress: Union[int, float], description: Optional[str] = None):
pass
+1 -1
View File
@@ -74,7 +74,7 @@ class ColumnFilter:
"""Identifies whether an item is found in the appropriate column"""
try:
if self.regex:
return re.search(self.pattern, f"{item}")
return bool(re.search(self.pattern, f"{item}"))
return self.pattern in f"{item}"
except OSError:
return False
+5 -5
View File
@@ -240,7 +240,7 @@ class Volshell(interfaces.plugins.PluginInterface):
return None
return self.context.modules[self.current_kernel_name]
def change_layer(self, layer_name: str = None):
def change_layer(self, layer_name: Optional[str] = None):
"""Changes the current default layer"""
if not layer_name:
layer_name = self.current_layer
@@ -250,7 +250,7 @@ class Volshell(interfaces.plugins.PluginInterface):
self.__current_layer = layer_name
sys.ps1 = f"({self.current_layer}) >>> "
def change_symbol_table(self, symbol_table_name: str = None):
def change_symbol_table(self, symbol_table_name: Optional[str] = None):
"""Changes the current_symbol_table"""
if not symbol_table_name:
print("No symbol table provided, not changing current symbol table")
@@ -262,7 +262,7 @@ class Volshell(interfaces.plugins.PluginInterface):
self.__current_symbol_table = symbol_table_name
print(f"Current Symbol Table: {self.current_symbol_table}")
def change_kernel(self, kernel_name: str = None):
def change_kernel(self, kernel_name: Optional[str] = None):
if not kernel_name:
print("No kernel module name provided, not changing current kernel")
if kernel_name not in self.context.modules:
@@ -347,7 +347,7 @@ class Volshell(interfaces.plugins.PluginInterface):
object: Union[
str, interfaces.objects.ObjectInterface, interfaces.objects.Template
],
offset: int = None,
offset: Optional[int] = None,
):
"""Display Type describes the members of a particular object in alphabetical order"""
if not isinstance(
@@ -479,7 +479,7 @@ class Volshell(interfaces.plugins.PluginInterface):
if treegrid is not None:
self.render_treegrid(treegrid)
def display_symbols(self, symbol_table: str = None):
def display_symbols(self, symbol_table: Optional[str] = None):
"""Prints an alphabetical list of symbols for a symbol table"""
if symbol_table is None:
print("No symbol table provided")
+3 -3
View File
@@ -2,7 +2,7 @@
# which is available at https://www.volatilityfoundation.org/license/vsl-v1.0
#
from typing import Any, List, Tuple, Union
from typing import Any, List, Optional, Tuple, Union
from volatility3.cli.volshell import generic
from volatility3.framework import constants, interfaces
@@ -61,7 +61,7 @@ class Volshell(generic.Volshell):
object: Union[
str, interfaces.objects.ObjectInterface, interfaces.objects.Template
],
offset: int = None,
offset: Optional[int] = None,
):
"""Display Type describes the members of a particular object in alphabetical order"""
if isinstance(object, str):
@@ -69,7 +69,7 @@ class Volshell(generic.Volshell):
object = self.current_symbol_table + constants.BANG + object
return super().display_type(object, offset)
def display_symbols(self, symbol_table: str = None):
def display_symbols(self, symbol_table: Optional[str] = None):
"""Prints an alphabetical list of symbols for a symbol table"""
if symbol_table is None:
symbol_table = self.current_symbol_table
+3 -3
View File
@@ -2,7 +2,7 @@
# which is available at https://www.volatilityfoundation.org/license/vsl-v1.0
#
from typing import Any, List, Tuple, Union
from typing import Any, List, Optional, Tuple, Union
from volatility3.cli.volshell import generic
from volatility3.framework import constants, interfaces
@@ -63,7 +63,7 @@ class Volshell(generic.Volshell):
object: Union[
str, interfaces.objects.ObjectInterface, interfaces.objects.Template
],
offset: int = None,
offset: Optional[int] = None,
):
"""Display Type describes the members of a particular object in alphabetical order"""
if isinstance(object, str):
@@ -71,7 +71,7 @@ class Volshell(generic.Volshell):
object = self.current_symbol_table + constants.BANG + object
return super().display_type(object, offset)
def display_symbols(self, symbol_table: str = None):
def display_symbols(self, symbol_table: Optional[str] = None):
"""Prints an alphabetical list of symbols for a symbol table"""
if symbol_table is None:
symbol_table = self.current_symbol_table
+3 -3
View File
@@ -2,7 +2,7 @@
# which is available at https://www.volatilityfoundation.org/license/vsl-v1.0
#
from typing import Any, List, Tuple, Union
from typing import Any, List, Optional, Tuple, Union
from volatility3.cli.volshell import generic
from volatility3.framework import constants, interfaces
@@ -60,7 +60,7 @@ class Volshell(generic.Volshell):
object: Union[
str, interfaces.objects.ObjectInterface, interfaces.objects.Template
],
offset: int = None,
offset: Optional[int] = None,
):
"""Display Type describes the members of a particular object in alphabetical order"""
if isinstance(object, str):
@@ -68,7 +68,7 @@ class Volshell(generic.Volshell):
object = self.current_symbol_table + constants.BANG + object
return super().display_type(object, offset)
def display_symbols(self, symbol_table: str = None):
def display_symbols(self, symbol_table: Optional[str] = None):
"""Prints an alphabetical list of symbols for a symbol table"""
if symbol_table is None:
symbol_table = self.current_symbol_table
+3 -4
View File
@@ -12,7 +12,7 @@ import inspect
import logging
import os
import traceback
from typing import Any, Dict, Generator, List, Tuple, Type, TypeVar
from typing import Any, Dict, Generator, List, Optional, Tuple, Type, TypeVar
from volatility3.framework import constants, interfaces
@@ -58,7 +58,7 @@ class NonInheritable:
self.default_value = value
self.cls = cls
def __get__(self, obj: Any, get_type: Type = None) -> Any:
def __get__(self, obj: Any, get_type: Type = Optional[None]) -> Any:
if type is self.cls:
if hasattr(self.default_value, "__get__"):
return self.default_value.__get__(obj, get_type)
@@ -185,8 +185,7 @@ def _zipwalk(path: str):
zip_results[os.path.join(path, os.path.dirname(file.filename))] = (
dirlist
)
for value in zip_results:
yield value, zip_results[value]
yield from zip_results.items()
def list_plugins() -> Dict[str, Type[interfaces.plugins.PluginInterface]]:
+3 -1
View File
@@ -166,7 +166,9 @@ class LayerStacker(interfaces.automagic.AutomagicInterface):
cls,
context: interfaces.context.ContextInterface,
initial_layer: str,
stack_set: List[Type[interfaces.automagic.StackerLayerInterface]] = None,
stack_set: Optional[
List[Type[interfaces.automagic.StackerLayerInterface]]
] = None,
progress_callback: constants.ProgressCallback = None,
):
"""Stacks as many possible layers on top of the initial layer as can be done.
@@ -104,9 +104,11 @@ class CacheManagerInterface(interfaces.configuration.VersionableInterface):
for subclazz in framework.class_subclasses(IdentifierProcessor):
self._classifiers[subclazz.operating_system] = subclazz
@abstractmethod
def add_identifier(self, location: str, operating_system: str, identifier: str):
"""Adds an identifier to the store"""
@abstractmethod
def find_location(
self, identifier: bytes, operating_system: Optional[str]
) -> Optional[str]:
@@ -120,15 +122,18 @@ class CacheManagerInterface(interfaces.configuration.VersionableInterface):
The location of the symbols file that matches the identifier
"""
@abstractmethod
def get_local_locations(self) -> Iterable[str]:
"""Returns a list of all the local locations"""
@abstractmethod
def update(self):
"""Locates all files under the symbol directories. Updates the cache with additions, modifications and removals.
This also updates remote locations based on a cache timeout.
"""
@abstractmethod
def get_identifier_dictionary(
self, operating_system: Optional[str] = None, local_only: bool = False
) -> Dict[bytes, str]:
@@ -142,12 +147,15 @@ class CacheManagerInterface(interfaces.configuration.VersionableInterface):
A dictionary of identifiers mapped to a location
"""
@abstractmethod
def get_identifier(self, location: str) -> Optional[bytes]:
"""Returns an identifier based on a specific location or None"""
@abstractmethod
def get_identifiers(self, operating_system: Optional[str]) -> List[bytes]:
"""Returns all identifiers for a particular operating system"""
@abstractmethod
def get_location_statistics(
self, location: str
) -> Optional[Tuple[int, int, int, int]]:
@@ -157,6 +165,7 @@ class CacheManagerInterface(interfaces.configuration.VersionableInterface):
A tuple of base_types, types, enums, symbols, or None is location not found
"""
@abstractmethod
def get_hash(self, location: str) -> Optional[str]:
"""Returns the hash of the JSON from within a location ISF"""
@@ -11,7 +11,7 @@ expect to be in the context (such as particular layers or symboltables).
import abc
import logging
import os
from typing import Any, ClassVar, Dict, List, Optional, Tuple, Type
from typing import Any, ClassVar, Dict, List, Optional, Set, Tuple, Type
from urllib import parse, request
from volatility3.framework import constants, interfaces
@@ -314,11 +314,11 @@ class TranslationLayerRequirement(
def __init__(
self,
name: str,
description: str = None,
description: Optional[str] = None,
default: interfaces.configuration.ConfigSimpleType = None,
optional: bool = False,
oses: List = None,
architectures: List = None,
oses: Optional[List] = None,
architectures: Optional[List[str]] = None,
) -> None:
"""Constructs a Translation Layer Requirement.
@@ -526,18 +526,18 @@ class VersionRequirement(interfaces.configuration.RequirementInterface):
description: Optional[str] = None,
default: bool = False,
optional: bool = False,
component: Type[interfaces.configuration.VersionableInterface] = None,
component: Optional[Type[interfaces.configuration.VersionableInterface]] = None,
version: Optional[Tuple[int, ...]] = None,
) -> None:
if version is None:
raise TypeError("Version cannot be None")
if component is None:
raise TypeError("Component cannot be None")
if description is None:
description = f"Version {'.'.join(str(x) for x in version)} dependency on {component.__module__}.{component.__name__} unmet"
super().__init__(
name=name, description=description, default=default, optional=optional
)
if component is None:
raise TypeError("Component cannot be None")
self._component: Type[interfaces.configuration.VersionableInterface] = component
self._version = version
@@ -546,7 +546,7 @@ class VersionRequirement(interfaces.configuration.RequirementInterface):
context: interfaces.context.ContextInterface,
config_path: str,
accumulator: Optional[
List[interfaces.configuration.VersionableInterface]
Set[interfaces.configuration.VersionableInterface]
] = None,
) -> Dict[str, interfaces.configuration.RequirementInterface]:
# Mypy doesn't appreciate our classproperty implementation, self._plugin.version has no type
@@ -580,7 +580,7 @@ class VersionRequirement(interfaces.configuration.RequirementInterface):
)
if result:
result.update({config_path: self})
result[config_path] = self
return result
context.config[interfaces.configuration.path_join(config_path, self.name)] = (
@@ -604,10 +604,10 @@ class PluginRequirement(VersionRequirement):
def __init__(
self,
name: str,
description: str = None,
description: Optional[str] = None,
default: bool = False,
optional: bool = False,
plugin: Type[interfaces.plugins.PluginInterface] = None,
plugin: Optional[Type[interfaces.plugins.PluginInterface]] = None,
version: Optional[Tuple[int, ...]] = None,
) -> None:
super().__init__(
@@ -627,7 +627,7 @@ class ModuleRequirement(
def __init__(
self,
name: str,
description: str = None,
description: Optional[str] = None,
default: bool = False,
architectures: Optional[List[str]] = None,
optional: bool = False,
+1 -1
View File
@@ -229,7 +229,7 @@ class Module(interfaces.context.ModuleInterface):
def object(
self,
object_type: str,
offset: int = None,
offset: Optional[int] = None,
native_layer_name: Optional[str] = None,
absolute: bool = False,
**kwargs,
@@ -42,7 +42,7 @@ class AutomagicInterface(
priority = 10
"""An ordering to indicate how soon this automagic should be run"""
exclusion_list = []
exclusion_list: List[str] = []
"""A list of plugin categories (typically operating systems) which the plugin will not operate on"""
def __init__(
@@ -82,7 +82,7 @@ class HierarchicalDict(collections.abc.Mapping):
def __init__(
self,
initial_dict: Dict[str, "SimpleTypeRequirement"] = None,
initial_dict: Optional[Dict[str, "SimpleTypeRequirement"]] = None,
separator: str = CONFIG_SEPARATOR,
) -> None:
"""
@@ -328,7 +328,7 @@ class RequirementInterface(metaclass=ABCMeta):
def __init__(
self,
name: str,
description: str = None,
description: Optional[str] = None,
default: ConfigSimpleType = None,
optional: bool = False,
) -> None:
@@ -618,7 +618,7 @@ class ConstructableRequirementInterface(RequirementInterface):
self,
context: "interfaces.context.ContextInterface",
config_path: str,
requirement_dict: Dict[str, object] = None,
requirement_dict: Optional[Dict[str, object]] = None,
) -> Optional["interfaces.objects.ObjectInterface"]:
"""Constructs the class, handing args and the subrequirements as
parameters to __init__"""
@@ -652,6 +652,7 @@ class ConstructableRequirementInterface(RequirementInterface):
class ConfigurableRequirementInterface(RequirementInterface):
"""Simple Abstract class to provide build_required_config."""
@abstractmethod
def build_configuration(
self,
context: "interfaces.context.ContextInterface",
+12 -2
View File
@@ -85,7 +85,7 @@ class ContextInterface(metaclass=ABCMeta):
object_type: Union[str, "interfaces.objects.Template"],
layer_name: str,
offset: int,
native_layer_name: str = None,
native_layer_name: Optional[str] = None,
**arguments,
) -> "interfaces.objects.ObjectInterface":
"""Object factory, takes a context, symbol, offset and optional
@@ -114,6 +114,7 @@ class ContextInterface(metaclass=ABCMeta):
"""
return copy.deepcopy(self)
@abstractmethod
def module(
self,
module_name: str,
@@ -232,7 +233,7 @@ class ModuleInterface(interfaces.configuration.ConfigurableInterface):
def object(
self,
object_type: str,
offset: int = None,
offset: Optional[int] = None,
native_layer_name: Optional[str] = None,
absolute: bool = False,
**kwargs,
@@ -277,27 +278,35 @@ class ModuleInterface(interfaces.configuration.ConfigurableInterface):
symbol = self.get_symbol(name)
return self.offset + symbol.address
@abstractmethod
def get_type(self, name: str) -> "interfaces.objects.Template":
"""Returns a type from the module's symbol table."""
@abstractmethod
def get_symbol(self, name: str) -> "interfaces.symbols.SymbolInterface":
"""Returns a symbol object from the module's symbol table."""
@abstractmethod
def get_enumeration(self, name: str) -> "interfaces.objects.Template":
"""Returns an enumeration from the module's symbol table."""
@abstractmethod
def has_type(self, name: str) -> bool:
"""Determines whether a type is present in the module's symbol table."""
@abstractmethod
def has_symbol(self, name: str) -> bool:
"""Determines whether a symbol is present in the module's symbol table."""
@abstractmethod
def has_enumeration(self, name: str) -> bool:
"""Determines whether an enumeration is present in the module's symbol table."""
@abstractmethod
def symbols(self) -> List:
"""Lists the symbols contained in the symbol table for this module"""
@abstractmethod
def get_symbols_by_absolute_location(self, offset: int, size: int = 0) -> List[str]:
"""Returns the symbols within table_name (or this module if not specified) that live at the specified
absolute offset provided."""
@@ -343,6 +352,7 @@ class ModuleContainer(collections.abc.Mapping):
def __iter__(self):
return iter(self._modules)
@abstractmethod
def free_module_name(self, prefix: str = "module") -> str:
"""Returns an unused table name to ensure no collision occurs when
inserting a symbol table."""
+1 -1
View File
@@ -210,7 +210,7 @@ class DataLayerInterface(
context: interfaces.context.ContextInterface,
scanner: ScannerInterface,
progress_callback: constants.ProgressCallback = None,
sections: Iterable[Tuple[int, int]] = None,
sections: Optional[Iterable[Tuple[int, int]]] = None,
) -> Iterable[Any]:
"""Scans a Translation layer by chunk.
@@ -374,6 +374,7 @@ class Template:
f"{self.__class__.__name__} object has no attribute {attr}"
)
@abc.abstractmethod
def __call__(
self,
context: "interfaces.context.ContextInterface",
@@ -183,7 +183,7 @@ class TreeGrid(metaclass=ABCMeta):
@abstractmethod
def populate(
self,
function: VisitorSignature = None,
function: Optional[VisitorSignature] = None,
initial_accumulator: Any = None,
fail_on_errors: bool = True,
) -> Optional[Exception]:
@@ -235,7 +235,7 @@ class TreeGrid(metaclass=ABCMeta):
node: Optional[TreeNode],
function: VisitorSignature,
initial_accumulator: _Type,
sort_key: ColumnSortKey = None,
sort_key: Optional[ColumnSortKey] = None,
) -> None:
"""Visits all the nodes in a tree, calling function on each one.
@@ -256,6 +256,7 @@ class SymbolSpaceInterface(collections.abc.Mapping):
"""An interface for the container that holds all the symbol-containing
tables for use within a context."""
@abstractmethod
def free_table_name(self, prefix: str = "layer") -> str:
"""Returns an unused table name to ensure no collision occurs when
inserting a symbol table."""
@@ -72,7 +72,7 @@ class MultiStringScanner(layers.ScannerInterface):
return None
for char in value:
trie[char] = trie.get(char, {})
trie.setdefault(char, {})
trie = trie[char]
# Mark the end of a string
+3 -3
View File
@@ -152,7 +152,7 @@ class PrimitiveObject(interfaces.objects.ObjectInterface):
type_name: str,
object_info: interfaces.objects.ObjectInformation,
data_format: DataFormatInfo,
new_value: TUnion[int, float, bool, bytes, str] = None,
new_value: Optional[TUnion[int, float, bool, bytes, str]] = None,
**kwargs,
) -> "PrimitiveObject":
"""Creates the appropriate class and returns it so that the native type
@@ -601,7 +601,7 @@ class Enumeration(interfaces.objects.ObjectInterface, int):
inverse_choices[v] = k
return inverse_choices
def lookup(self, value: int = None) -> str:
def lookup(self, value: Optional[int] = None) -> str:
"""Looks up an individual value and returns the associated name.
If multiple identifiers map to the same value, the first matching identifier will be returned
@@ -690,7 +690,7 @@ class Array(interfaces.objects.ObjectInterface, collections.abc.Sequence):
type_name: str,
object_info: interfaces.objects.ObjectInformation,
count: int = 0,
subtype: templates.ObjectTemplate = None,
subtype: Optional[templates.ObjectTemplate] = None,
) -> None:
super().__init__(context=context, type_name=type_name, object_info=object_info)
self._vol["count"] = count
@@ -2,7 +2,7 @@
# which is available at https://www.volatilityfoundation.org/license/vsl-v1.0
#
import datetime
from typing import Any, Callable, Iterable, List, Tuple
from typing import Any, Callable, Iterable, List, Optional, Tuple
from volatility3.framework import interfaces, renderers
from volatility3.framework.configuration import requirements
@@ -58,7 +58,9 @@ class PsList(interfaces.plugins.PluginInterface, timeliner.TimeLinerInterface):
]
@classmethod
def create_pid_filter(cls, pid_list: List[int] = None) -> Callable[[Any], bool]:
def create_pid_filter(
cls, pid_list: Optional[List[int]] = None
) -> Callable[[Any], bool]:
"""Constructs a filter function for process IDs.
Args:
+4 -2
View File
@@ -4,7 +4,7 @@
import datetime
import logging
from typing import Callable, Dict, Iterable, List
from typing import Callable, Dict, Iterable, List, Optional
from volatility3.framework import exceptions, interfaces, renderers
from volatility3.framework.configuration import requirements
@@ -82,7 +82,9 @@ class PsList(interfaces.plugins.PluginInterface):
return list_tasks
@classmethod
def create_pid_filter(cls, pid_list: List[int] = None) -> Callable[[int], bool]:
def create_pid_filter(
cls, pid_list: Optional[List[int]] = None
) -> Callable[[int], bool]:
def filter_func(_):
return False
+3 -1
View File
@@ -54,7 +54,9 @@ class Timeliner(interfaces.plugins.PluginInterface):
self.automagics: Optional[List[interfaces.automagic.AutomagicInterface]] = None
@classmethod
def get_usable_plugins(cls, selected_list: List[str] = None) -> List[Type]:
def get_usable_plugins(
cls, selected_list: Optional[List[str]] = None
) -> List[Type]:
# Initialize for the run
plugin_list = list(framework.class_subclasses(TimeLinerInterface))
@@ -2,7 +2,7 @@
# which is available at https://www.volatilityfoundation.org/license/vsl-v1.0
#
import logging
from typing import Generator, Iterable, List
from typing import Generator, Iterable, List, Optional
from volatility3.framework import constants, exceptions, interfaces, renderers
from volatility3.framework.configuration import requirements
@@ -133,7 +133,7 @@ class Modules(interfaces.plugins.PluginInterface):
context: interfaces.context.ContextInterface,
layer_name: str,
symbol_table: str,
pids: List[int] = None,
pids: Optional[List[int]] = None,
) -> Generator[str, None, None]:
"""Build a cache of possible virtual layers, in priority starting with
the primary/kernel layer. Then keep one layer per session by cycling
@@ -96,7 +96,7 @@ class PEDump(interfaces.plugins.PluginInterface):
pe_table_name: str,
ldr_entry: interfaces.objects.ObjectInterface,
open_method: Type[interfaces.plugins.FileHandlerInterface],
layer_name: str = None,
layer_name: Optional[str] = None,
prefix: str = "",
) -> Optional[str]:
"""Extracts the PE file referenced an LDR_DATA_TABLE_ENTRY (DLL, kernel module) instance
@@ -183,7 +183,7 @@ class PoolScanner(plugins.PluginInterface):
@staticmethod
def builtin_constraints(
symbol_table: str, tags_filter: List[bytes] = None
symbol_table: str, tags_filter: Optional[List[bytes]] = None
) -> List[PoolConstraint]:
"""Get built-in PoolConstraints given a list of pool tags.
@@ -4,7 +4,7 @@
import datetime
import logging
from typing import Callable, Iterator, List, Type
from typing import Callable, Iterator, List, Optional, Type
from volatility3.framework import renderers, interfaces, layers, exceptions, constants
from volatility3.framework.configuration import requirements
@@ -114,7 +114,7 @@ class PsList(interfaces.plugins.PluginInterface, timeliner.TimeLinerInterface):
@classmethod
def create_pid_filter(
cls, pid_list: List[int] = None, exclude: bool = False
cls, pid_list: Optional[List[int]] = None, exclude: bool = False
) -> Callable[[interfaces.objects.ObjectInterface], bool]:
"""A factory for producing filter functions that filter based on a list
of process IDs.
@@ -171,7 +171,7 @@ class PsList(interfaces.plugins.PluginInterface, timeliner.TimeLinerInterface):
@classmethod
def create_name_filter(
cls, name_list: List[str] = None, exclude: bool = False
cls, name_list: Optional[List[str]] = None, exclude: bool = False
) -> Callable[[interfaces.objects.ObjectInterface], bool]:
"""A factory for producing filter functions that filter based on a list
of process names.
@@ -89,7 +89,7 @@ class PsScan(interfaces.plugins.PluginInterface, timeliner.TimeLinerInterface):
cls,
context: interfaces.context.ContextInterface,
layer_name: str,
offset: int = None,
offset: Optional[int] = None,
physical: bool = True,
exclude: bool = False,
) -> Callable[[interfaces.objects.ObjectInterface], bool]:
@@ -4,7 +4,7 @@
import datetime
import logging
from typing import List, Sequence, Iterable, Tuple, Union
from typing import List, Optional, Sequence, Iterable, Tuple, Union
from volatility3.framework import objects, renderers, exceptions, interfaces, constants
from volatility3.framework.configuration import requirements
@@ -51,7 +51,7 @@ class PrintKey(interfaces.plugins.PluginInterface):
def key_iterator(
cls,
hive: RegistryHive,
node_path: Sequence[objects.StructType] = None,
node_path: Optional[Sequence[objects.StructType]] = None,
recurse: bool = False,
) -> Iterable[
Tuple[
@@ -121,7 +121,7 @@ class PrintKey(interfaces.plugins.PluginInterface):
def _printkey_iterator(
self,
hive: RegistryHive,
node_path: Sequence[objects.StructType] = None,
node_path: Optional[Sequence[objects.StructType]] = None,
recurse: bool = False,
):
"""Method that wraps the more generic key_iterator, to provide output
@@ -242,8 +242,8 @@ class PrintKey(interfaces.plugins.PluginInterface):
self,
layer_name: str,
symbol_table: str,
hive_offsets: List[int] = None,
key: str = None,
hive_offsets: Optional[List[int]] = None,
key: Optional[str] = None,
recurse: bool = False,
):
for hive in hivelist.HiveList.list_hives(
@@ -270,7 +270,6 @@ class _ScheduledTasksReader(io.BytesIO):
return val
def read_aligned_bstring_expand_sz(self) -> Optional[str]:
# type: () -> Optional[str]
sz = self.read_aligned_u4()
if sz is None:
return None
+1 -1
View File
@@ -214,7 +214,7 @@ class TreeGrid(interfaces.renderers.TreeGrid):
def populate(
self,
function: interfaces.renderers.VisitorSignature = None,
function: Optional[interfaces.renderers.VisitorSignature] = None,
initial_accumulator: Any = None,
fail_on_errors: bool = True,
) -> Optional[Exception]:
+4 -4
View File
@@ -53,10 +53,10 @@ class SymbolSpace(interfaces.symbols.SymbolSpaceInterface):
self._resolved: Dict[str, interfaces.objects.Template] = {}
self._resolved_symbols: Dict[str, interfaces.objects.Template] = {}
def clear_symbol_cache(self, table_name: str = None) -> None:
def clear_symbol_cache(self, table_name: Optional[str] = None) -> None:
"""Clears the symbol cache for the specified table name. If no table
name is specified, the caches of all symbol tables are cleared."""
table_list: List[interfaces.symbols.BaseSymbolTableInterface] = list()
table_list: List[interfaces.symbols.BaseSymbolTableInterface] = []
if table_name is None:
table_list = list(self._dict.values())
else:
@@ -81,7 +81,7 @@ class SymbolSpace(interfaces.symbols.SymbolSpaceInterface):
yield table + constants.BANG + symbol_name
def get_symbols_by_location(
self, offset: int, size: int = 0, table_name: str = None
self, offset: int, size: int = 0, table_name: Optional[str] = None
) -> Iterable[str]:
"""Returns all symbols that exist at a specific relative address."""
table_list: Iterable[interfaces.symbols.BaseSymbolTableInterface] = (
@@ -128,7 +128,7 @@ class SymbolSpace(interfaces.symbols.SymbolSpaceInterface):
self,
producer: str,
validator: Callable[[Optional[Tuple], Optional[datetime.datetime]], bool],
tables: List[str] = None,
tables: Optional[List[str]] = None,
) -> bool:
"""Verifies the producer metadata and version of tables
@@ -4,7 +4,7 @@
import random
import string
from typing import Union
from typing import Optional, Union
from volatility3.framework import objects, interfaces
@@ -14,8 +14,8 @@ class GenericIntelProcess(objects.StructType):
self,
context: interfaces.context.ContextInterface,
dtb: Union[int, interfaces.objects.ObjectInterface],
config_prefix: str = None,
preferred_name: str = None,
config_prefix: Optional[str] = None,
preferred_name: Optional[str] = None,
) -> str:
"""Constructs a new layer based on the process's DirectoryTableBase."""
+2 -2
View File
@@ -86,7 +86,7 @@ class IntermediateSymbolTable(interfaces.symbols.SymbolTableInterface):
config_path: str,
name: str,
isf_url: str,
native_types: interfaces.symbols.NativeTableInterface = None,
native_types: Optional[interfaces.symbols.NativeTableInterface] = None,
table_mapping: Optional[Dict[str, str]] = None,
validate: bool = True,
class_types: Optional[
@@ -319,7 +319,7 @@ class ISFormatTable(interfaces.symbols.SymbolTableInterface, metaclass=ABCMeta):
config_path: str,
name: str,
json_object: Any,
native_types: interfaces.symbols.NativeTableInterface = None,
native_types: Optional[interfaces.symbols.NativeTableInterface] = None,
table_mapping: Optional[Dict[str, str]] = None,
) -> None:
self._json_object = json_object
@@ -308,7 +308,7 @@ class module(generic.GenericIntelProcess):
class task_struct(generic.GenericIntelProcess):
def add_process_layer(
self, config_prefix: str = None, preferred_name: str = None
self, config_prefix: Optional[str] = None, preferred_name: Optional[str] = None
) -> Optional[str]:
"""Constructs a new layer based on the process's DTB.
@@ -1,7 +1,7 @@
# 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 typing import Iterator, Any, Iterable, List, Optional, Tuple, Set
from volatility3.framework import interfaces, objects, exceptions, constants
from volatility3.framework.symbols import intermed
@@ -97,7 +97,7 @@ class MacUtilities(interfaces.configuration.VersionableInterface):
context: interfaces.context.ContextInterface,
handlers: Iterator[Any],
target_address,
kernel_module_name: str = None,
kernel_module_name: Optional[str] = None,
):
mod_name = "UNKNOWN"
symbol_name = "N/A"
@@ -18,7 +18,7 @@ class proc(generic.GenericIntelProcess):
return self.task.dereference().cast("task")
def add_process_layer(
self, config_prefix: str = None, preferred_name: str = None
self, config_prefix: Optional[str] = None, preferred_name: Optional[str] = None
) -> Optional[str]:
"""Constructs a new layer based on the process's DTB.
+1 -1
View File
@@ -25,7 +25,7 @@ class ProducerMetadata(interfaces.symbols.MetadataInterface):
return self._json_data.get("version", "")
@property
def version(self) -> Optional[Tuple[int]]:
def version(self) -> Optional[Tuple[int, ...]]:
"""Returns the version of the ISF file producer"""
version = self.version_string
if not version:
@@ -692,7 +692,9 @@ class EPROCESS(generic.GenericIntelProcess, pool.ExecutiveObject):
return True
def add_process_layer(self, config_prefix: str = None, preferred_name: str = None):
def add_process_layer(
self, config_prefix: Optional[str] = None, preferred_name: Optional[str] = None
):
"""Constructs a new layer based on the process's DirectoryTableBase."""
parent_layer = self._context.layers[self.vol.layer_name]
@@ -362,7 +362,7 @@ class OBJECT_HEADER(objects.StructType):
return True
def get_object_type(
self, type_map: Dict[int, str], cookie: int = None
self, type_map: Dict[int, str], cookie: Optional[int] = None
) -> Optional[str]:
"""Across all Windows versions, the _OBJECT_HEADER embeds details on
the type of object (i.e. process, file) but the way its embedded
@@ -984,7 +984,9 @@ if __name__ == "__main__":
def __init__(self):
self._max_message_len = 0
def __call__(self, progress: Union[int, float], description: str = None):
def __call__(
self, progress: Union[int, float], description: Optional[str] = None
):
"""A simple function for providing text-based feedback.
.. warning:: Only for development use.
@@ -36,7 +36,7 @@ class PDBUtility(interfaces.configuration.VersionableInterface):
layer_name: str,
offset: int,
symbol_table_class: str = "volatility3.framework.symbols.intermed.IntermediateSymbolTable",
config_path: str = None,
config_path: Optional[str] = None,
progress_callback: constants.ProgressCallback = None,
) -> Optional[str]:
"""Produces the name of a symbol table loaded from the offset for an MZ header
@@ -388,8 +388,8 @@ class PDBUtility(interfaces.configuration.VersionableInterface):
config_path: str,
layer_name: str,
pdb_name: str,
module_offset: int = None,
module_size: int = None,
module_offset: Optional[int] = None,
module_size: Optional[int] = None,
) -> str:
"""Creates symbol table for a module in the specified layer_name.
@@ -418,8 +418,8 @@ class PDBUtility(interfaces.configuration.VersionableInterface):
config_path: str,
layer_name: str,
pdb_name: str,
module_offset: int = None,
module_size: int = None,
module_offset: Optional[int] = None,
module_size: Optional[int] = None,
create_module: bool = False,
) -> Tuple[Optional[str], Optional[str]]:
if module_offset is None:
@@ -478,8 +478,8 @@ class PDBUtility(interfaces.configuration.VersionableInterface):
config_path: str,
layer_name: str,
pdb_name: str,
module_offset: int = None,
module_size: int = None,
module_offset: Optional[int] = None,
module_size: Optional[int] = None,
) -> str:
"""Creates a module in the specified layer_name based on a pdb name.