mirror of
https://github.com/volatilityfoundation/volatility3.git
synced 2026-09-10 19:57:39 +02:00
Tidy up exceptions to be more accurate.
This commit is contained in:
@@ -393,7 +393,8 @@ class CommandLine(interfaces.plugins.FileConsumerInterface):
|
||||
if isinstance(value, str):
|
||||
if not parse.urlparse(value).scheme:
|
||||
if not os.path.exists(value):
|
||||
raise TypeError("Non-existant file {} passed to URIRequirement".format(value))
|
||||
raise FileNotFoundError(
|
||||
"Non-existant file {} passed to URIRequirement".format(value))
|
||||
value = "file://" + request.pathname2url(os.path.abspath(value))
|
||||
if isinstance(requirement, requirements.ListRequirement):
|
||||
if not isinstance(value, list):
|
||||
@@ -410,7 +411,7 @@ class CommandLine(interfaces.plugins.FileConsumerInterface):
|
||||
def consume_file(self, filedata: interfaces.plugins.FileInterface):
|
||||
"""Consumes a file as produced by a plugin."""
|
||||
if self.output_dir is None:
|
||||
raise ValueError("Output directory has not been correctly specified")
|
||||
raise TypeError("Output directory is not a string")
|
||||
os.makedirs(self.output_dir, exist_ok = True)
|
||||
|
||||
pref_name_array = filedata.preferred_filename.split('.')
|
||||
|
||||
@@ -103,7 +103,7 @@ class ListRequirement(interfaces.configuration.RequirementInterface):
|
||||
return {config_path: self}
|
||||
if not isinstance(value, list):
|
||||
# TODO: Check this is the correct response for an error
|
||||
raise ValueError("Unexpected config value found: {}".format(repr(value)))
|
||||
raise TypeError("Unexpected config value found: {}".format(repr(value)))
|
||||
if not (self.min_elements <= len(value)):
|
||||
vollog.log(constants.LOGLEVEL_V, "TypeError - Too few values provided to list option.")
|
||||
return {config_path: self}
|
||||
@@ -352,7 +352,7 @@ class SymbolTableRequirement(interfaces.configuration.ConstructableRequirementIn
|
||||
|
||||
# Fill out the parameter for class creation
|
||||
if not isinstance(self.requirements["class"], interfaces.configuration.ClassRequirement):
|
||||
raise ValueError("Class requirement is not of type ClassRequirement: {}".format(
|
||||
raise TypeError("Class requirement is not of type ClassRequirement: {}".format(
|
||||
repr(self.requirements["class"])))
|
||||
cls = self.requirements["class"].cls
|
||||
node_config = context.config.branch(config_path)
|
||||
@@ -383,10 +383,10 @@ class PluginRequirement(interfaces.configuration.RequirementInterface):
|
||||
version: Optional[Tuple[int, ...]] = None) -> None:
|
||||
super().__init__(name = name, description = description, default = default, optional = optional)
|
||||
if plugin is None:
|
||||
raise ValueError("Plugin cannot be None")
|
||||
raise TypeError("Plugin cannot be None")
|
||||
self._plugin = plugin
|
||||
if version is None:
|
||||
raise ValueError("Version cannot be None")
|
||||
raise TypeError("Version cannot be None")
|
||||
self._version = version
|
||||
|
||||
def unsatisfied(self, context: interfaces.context.ContextInterface,
|
||||
|
||||
@@ -184,7 +184,7 @@ class Module(interfaces.context.ModuleInterface):
|
||||
raise ValueError("Cannot reference another module when constructing an object")
|
||||
|
||||
if offset is None:
|
||||
raise ValueError("Offset must not be None for non-symbol objects")
|
||||
raise TypeError("Offset must not be None for non-symbol objects")
|
||||
|
||||
if not absolute:
|
||||
offset += self._offset
|
||||
@@ -227,7 +227,7 @@ class Module(interfaces.context.ModuleInterface):
|
||||
offset += self._offset
|
||||
|
||||
if symbol_val.type is None:
|
||||
raise ValueError("Symbol {} has no associated type".format(symbol_val.name))
|
||||
raise TypeError("Symbol {} has no associated type".format(symbol_val.name))
|
||||
|
||||
# Ensure we don't use a layer_name other than the module's, why would anyone do that?
|
||||
if 'layer_name' in kwargs:
|
||||
|
||||
@@ -42,7 +42,7 @@ class AutomagicInterface(interfaces.configuration.ConfigurableInterface, metacla
|
||||
for requirement in self.get_requirements():
|
||||
if not isinstance(requirement, (interfaces.configuration.SimpleTypeRequirement,
|
||||
requirements.ChoiceRequirement, requirements.ListRequirement)):
|
||||
raise ValueError(
|
||||
raise TypeError(
|
||||
"Automagic requirements must be a SimpleTypeRequirement, ChoiceRequirement or ListRequirement")
|
||||
|
||||
def __call__(self,
|
||||
|
||||
@@ -11,6 +11,10 @@ from volatility.framework.objects import utility
|
||||
from volatility.framework.symbols import intermed
|
||||
|
||||
|
||||
class PDBFormatException(exceptions.LayerException):
|
||||
"""Thrown when an error occurs with the underlying MSF file format."""
|
||||
|
||||
|
||||
class PdbMultiStreamFormat(linear.LinearlyMappedLayer):
|
||||
_headers = {
|
||||
"MSF_HDR": "Microsoft C/C++ program database 2.00\r\n\x1a\x4a\x47",
|
||||
@@ -28,7 +32,7 @@ class PdbMultiStreamFormat(linear.LinearlyMappedLayer):
|
||||
self._pdb_symbol_table = intermed.IntermediateSymbolTable.create(context, self._config_path, 'windows', 'pdb')
|
||||
response = self._check_header()
|
||||
if response is None:
|
||||
raise ValueError("Could not find a suitable header")
|
||||
raise PDBFormatException(name, "Could not find a suitable header")
|
||||
self._version, self._header = response
|
||||
self._streams = {} # type: Dict[int, str]
|
||||
|
||||
@@ -138,7 +142,7 @@ class PdbMultiStreamFormat(linear.LinearlyMappedLayer):
|
||||
def get_stream(self, index) -> Optional['PdbMSFStream']:
|
||||
self.read_streams()
|
||||
if index not in self._streams:
|
||||
raise ValueError("Stream not present")
|
||||
raise PDBFormatException(self.name, "Stream not present")
|
||||
if self._streams[index]:
|
||||
layer = self.context.layers[self._streams[index]]
|
||||
if isinstance(layer, PdbMSFStream):
|
||||
@@ -158,7 +162,7 @@ class PdbMSFStream(linear.LinearlyMappedLayer):
|
||||
self._pages = self.config.get("pages", None)
|
||||
self._pages_len = len(self._pages)
|
||||
if not self._pages:
|
||||
raise ValueError("Invalid/no pages specified")
|
||||
raise PDBFormatException(name, "Invalid/no pages specified")
|
||||
if not isinstance(self._pdb_layer, PdbMultiStreamFormat):
|
||||
raise TypeError("Base Layer must be a PdbMultiStreamFormat layer")
|
||||
|
||||
@@ -212,8 +216,9 @@ class PdbMSFStream(linear.LinearlyMappedLayer):
|
||||
@property
|
||||
def _pdb_layer(self) -> PdbMultiStreamFormat:
|
||||
if self._base_layer not in self._context.layers:
|
||||
raise ValueError("No PdbMultiStreamFormat layer found: {}".format(self._base_layer))
|
||||
raise PDBFormatException(self._base_layer,
|
||||
"No PdbMultiStreamFormat layer found: {}".format(self._base_layer))
|
||||
result = self._context.layers[self._base_layer]
|
||||
if isinstance(result, PdbMultiStreamFormat):
|
||||
return result
|
||||
raise ValueError("Base layer is not PdbMultiStreamFormat")
|
||||
raise TypeError("Base layer is not PdbMultiStreamFormat")
|
||||
|
||||
@@ -130,7 +130,7 @@ class FileLayer(interfaces.layers.DataLayerInterface):
|
||||
def is_valid(self, offset: int, length: int = 1) -> bool:
|
||||
"""Returns whether the offset is valid or not."""
|
||||
if length <= 0:
|
||||
raise TypeError("Length must be positive")
|
||||
raise ValueError("Length must be positive")
|
||||
return bool(self.minimum_address <= offset <= self.maximum_address
|
||||
and self.minimum_address <= offset + length - 1 <= self.maximum_address)
|
||||
|
||||
|
||||
@@ -5,12 +5,16 @@
|
||||
import struct
|
||||
from typing import Any, Dict, List, Optional
|
||||
|
||||
from volatility.framework import interfaces, constants
|
||||
from volatility.framework import interfaces, constants, exceptions
|
||||
from volatility.framework.configuration import requirements
|
||||
from volatility.framework.layers import physical, segmented, resources
|
||||
from volatility.framework.symbols import native
|
||||
|
||||
|
||||
class VmwareFormatException(exceptions.LayerException):
|
||||
"""Thrown when an error occurs with the underlying Crash file format."""
|
||||
|
||||
|
||||
class VmwareLayer(segmented.SegmentedLayer):
|
||||
priority = 22
|
||||
|
||||
@@ -44,7 +48,7 @@ class VmwareLayer(segmented.SegmentedLayer):
|
||||
data = meta_layer.read(0, header_size)
|
||||
magic, unknown, groupCount = struct.unpack(self.header_structure, data)
|
||||
if magic not in [b"\xD2\xBE\xD2\xBE"]:
|
||||
raise ValueError("Wrong magic bytes for Vmware layer: {}".format(repr(magic)))
|
||||
raise VmwareFormatException(self.name, "Wrong magic bytes for Vmware layer: {}".format(repr(magic)))
|
||||
|
||||
# TODO: Change certain structure sizes based on the version
|
||||
version = magic[1] & 0xf
|
||||
@@ -87,7 +91,7 @@ class VmwareLayer(segmented.SegmentedLayer):
|
||||
index_len) + self._context.symbol_space.get_type("vmware!unsigned int").size
|
||||
|
||||
if tags[("regionsCount", ())][1] == 0:
|
||||
raise ValueError("VMware VMEM is not split into regions")
|
||||
raise VmwareFormatException(self.name, "VMware VMEM is not split into regions")
|
||||
for region in range(tags[("regionsCount", ())][1]):
|
||||
offset = tags[("regionPPN", (region, ))][1] * self._page_size
|
||||
mapped_offset = tags[("regionPageNum", (region, ))][1] * self._page_size
|
||||
|
||||
@@ -26,7 +26,7 @@ def convert_data_to_value(data: bytes, struct_type: Type[TUnion[int, float, byte
|
||||
elif struct_type == float:
|
||||
float_vals = "zzezfzzzd"
|
||||
if data_format.length > len(float_vals) or float_vals[data_format.length] not in "efd":
|
||||
raise TypeError("Invalid float size")
|
||||
raise ValueError("Invalid float size")
|
||||
struct_format = ("<" if data_format.byteorder == 'little' else ">") + float_vals[data_format.length]
|
||||
elif struct_type in [bytes, str]:
|
||||
struct_format = str(data_format.length) + "s"
|
||||
@@ -54,7 +54,7 @@ def convert_value_to_data(value: TUnion[int, float, bytes, str, bool],
|
||||
elif struct_type == float:
|
||||
float_vals = "zzezfzzzd"
|
||||
if data_format.length > len(float_vals) or float_vals[data_format.length] not in "efd":
|
||||
raise TypeError("Invalid float size")
|
||||
raise ValueError("Invalid float size")
|
||||
struct_format = ("<" if data_format.byteorder == 'little' else ">") + float_vals[data_format.length]
|
||||
elif struct_type in [bytes, str]:
|
||||
struct_format = str(data_format.length) + "s"
|
||||
@@ -279,7 +279,7 @@ class Pointer(Integer):
|
||||
"""
|
||||
length, endian, signed = data_format
|
||||
if signed:
|
||||
raise TypeError("Pointers cannot have signed values")
|
||||
raise ValueError("Pointers cannot have signed values")
|
||||
mask = context.layers[object_info.native_layer_name].address_mask
|
||||
data = context.layers.read(object_info.layer_name, object_info.offset, length)
|
||||
value = int.from_bytes(data, byteorder = endian, signed = signed)
|
||||
@@ -514,7 +514,7 @@ class Array(interfaces.objects.ObjectInterface, abc.Sequence):
|
||||
"""Returns the size of the array, based on the count and the
|
||||
subtype."""
|
||||
if 'subtype' not in template.vol and 'count' not in template.vol:
|
||||
raise TypeError("Array ObjectTemplate must be provided a count and subtype")
|
||||
raise ValueError("Array ObjectTemplate must be provided a count and subtype")
|
||||
return template.vol.get('subtype', None).size * template.vol.get('count', 0)
|
||||
|
||||
@classmethod
|
||||
@@ -606,7 +606,7 @@ class AggregateType(interfaces.objects.ObjectInterface):
|
||||
def size(cls, template: interfaces.objects.Template) -> int:
|
||||
"""Method to return the size of this type."""
|
||||
if template.vol.get('size', None) is None:
|
||||
raise TypeError("ObjectTemplate not provided with a size")
|
||||
raise ValueError("ObjectTemplate not provided with a size")
|
||||
return template.vol.size
|
||||
|
||||
@classmethod
|
||||
|
||||
@@ -153,9 +153,6 @@ def os_distinguisher(version_check: Callable[[Tuple[int, ...]], bool],
|
||||
except (AttributeError, ValueError, TypeError):
|
||||
vollog.log(constants.LOGLEVEL_VVV, "Windows PE version data is not available")
|
||||
|
||||
if not fallback_checks:
|
||||
raise ValueError("No fallback methods for os_distinguishing provided")
|
||||
|
||||
# fall back to the backup method, if necessary
|
||||
for name, member, response in fallback_checks:
|
||||
if member is None:
|
||||
|
||||
@@ -52,7 +52,7 @@ class VerInfo(interfaces.plugins.PluginInterface):
|
||||
"""
|
||||
|
||||
if layer_name is None:
|
||||
raise ValueError("Layer must be a string not None")
|
||||
raise TypeError("Layer must be a string not None")
|
||||
|
||||
pe_data = io.BytesIO()
|
||||
|
||||
|
||||
@@ -243,7 +243,7 @@ class TreeGrid(interfaces.renderers.TreeGrid):
|
||||
The values returned are mutable,
|
||||
"""
|
||||
if node is None:
|
||||
raise ValueError("Node must be a valid node within the TreeGrid")
|
||||
raise TypeError("Node must be a valid node within the TreeGrid")
|
||||
return node.values
|
||||
|
||||
def _append(self, parent, values):
|
||||
|
||||
@@ -118,7 +118,7 @@ class SymbolSpace(interfaces.symbols.SymbolSpaceInterface):
|
||||
elif resolve_type == SymbolType.ENUM:
|
||||
get_function = 'get_enumeration'
|
||||
else:
|
||||
raise ValueError("Weak_resolve called without a proper SymbolType")
|
||||
raise TypeError("Weak_resolve called without a proper SymbolType")
|
||||
|
||||
name_array = name.split(constants.BANG)
|
||||
if len(name_array) == 2:
|
||||
|
||||
@@ -224,7 +224,7 @@ class IntermediateSymbolTable(interfaces.symbols.SymbolTableInterface):
|
||||
"""
|
||||
urls = list(cls.file_symbol_url(sub_path, filename))
|
||||
if not urls:
|
||||
raise ValueError("No symbol files found at provided filename: {}", filename)
|
||||
raise FileNotFoundError("No symbol files found at provided filename: {}", filename)
|
||||
table_name = context.symbol_space.free_table_name(filename)
|
||||
table = cls(context = context,
|
||||
config_path = config_path,
|
||||
@@ -260,7 +260,7 @@ class ISFormatTable(interfaces.symbols.SymbolTableInterface, metaclass = ABCMeta
|
||||
self.name = name
|
||||
nt = native_types or self._get_natives()
|
||||
if nt is None:
|
||||
raise ValueError("Native table not provided")
|
||||
raise TypeError("Native table not provided")
|
||||
nt.name = name + "_natives"
|
||||
super().__init__(context, config_path, name, nt, table_mapping = table_mapping)
|
||||
self._overrides = {} # type: Dict[str, Type[interfaces.objects.ObjectInterface]]
|
||||
|
||||
@@ -111,7 +111,7 @@ class SERVICE_RECORD(objects.StructType):
|
||||
yield rec
|
||||
rec = rec.ServiceList.Blink.dereference()
|
||||
except exceptions.InvalidAddressException:
|
||||
raise StopIteration
|
||||
return
|
||||
|
||||
|
||||
class SERVICE_HEADER(objects.StructType):
|
||||
|
||||
@@ -362,7 +362,7 @@ class PdbReader:
|
||||
self._progress_callback(offset * 100 / tpi_layer.maximum_address, "Reading TPI layer")
|
||||
length = module.object(object_type = length_type, offset = offset)
|
||||
if not isinstance(length, int):
|
||||
raise ValueError("Non-integer length provided")
|
||||
raise TypeError("Non-integer length provided")
|
||||
offset += length_len
|
||||
output, consumed = self.consume_type(module, offset, length)
|
||||
leaf_type, name, value = output
|
||||
@@ -783,7 +783,7 @@ class PdbReader:
|
||||
result = leaf_type, None, bitfield
|
||||
consumed += remaining
|
||||
else:
|
||||
raise ValueError("Unhandled leaf_type: {}".format(leaf_type))
|
||||
raise TypeError("Unhandled leaf_type: {}".format(leaf_type))
|
||||
|
||||
return result, consumed
|
||||
|
||||
|
||||
Reference in New Issue
Block a user