diff --git a/volatility3/framework/objects/utility.py b/volatility3/framework/objects/utility.py index 1fcd305f7..a1ad4fdf5 100644 --- a/volatility3/framework/objects/utility.py +++ b/volatility3/framework/objects/utility.py @@ -2,10 +2,9 @@ # which is available at https://www.volatilityfoundation.org/license/vsl-v1.0 # -import re from typing import Optional, Union -from volatility3.framework import interfaces, objects, constants +from volatility3.framework import interfaces, objects, constants, exceptions def rol(value: int, count: int, max_bits: int = 64) -> int: @@ -102,6 +101,64 @@ def pointer_to_string( ) +def gather_contiguous_bytes_from_address(layer, address: int, count: int) -> bytes: + """ + This method reconstructs a string from memory while also carefully examining each page + + It goes page-by-page reading the bytes. This is done by calculating page boundaries + and then only reading one page at a time. + + If a page is missing, the code initially catches the exception. + If data is non-empty (meaning at least one read succeeded), then we return what was read + If the first page fails, then we re-raise the exception + """ + + data = b"" + + left_to_read = count + + # read as many pages as possible that are contiguous + # if the first page is missed, we re-raise the InvalidAddressException + # if we have at least 1 page that was read succesfully, + # then we try to construct a string from it + while left_to_read > 0: + # compute aligned address of current page and next the page + aligned = address & ~0xFFF + next_page = aligned + 0xFFF + 1 + + # all fits on the current page, last read + if address + left_to_read < next_page: + try: + data += layer.read(address, left_to_read) + except exceptions.InvalidAddressException: + # if we have data, just break the loop + if data: + break + # Raise if no data was read as this means the first page was invalid + else: + raise + + left_to_read = 0 + + else: + # how many bytes are left on the current page + len_to_read = next_page - address + + try: + data += layer.read(address, len_to_read) + except exceptions.InvalidAddressException: + if data: + break + # Raise if no data was read as this means the first page was invalid + else: + raise + + address += len_to_read + left_to_read -= len_to_read + + return data + + def address_to_string( context: interfaces.context.ContextInterface, layer_name: str, @@ -131,18 +188,38 @@ def address_to_string( if count < 1: raise ValueError("Count must be greater than 0") + encodings = {"utf8": 1, "utf16": 2, "utf32": 4} + if encoding not in encodings: + raise ValueError( + f"Encoding ({encoding} is invalid. Must be one of {[e for e in encodings]}." + ) + layer = context.layers[layer_name] - # Purposely do not catch exception - data = layer.read(address, count) + data = gather_contiguous_bytes_from_address(layer, address, count) - decoded_data = data.decode(encoding=encoding, errors=errors) - try: - idx = re.search("\ufffd|\x00", decoded_data).start() - except AttributeError: - idx = len(decoded_data) + # we need to find the ending nulls, which the amount of nulls varies based on encoding + ending_nulls = b"\x00" * encodings[encoding] - return decoded_data[:idx] + end_idx = data.find(ending_nulls) + # send back the bytes even if the ending nulls aren't found (can be on the next page) + if end_idx == -1: + return data + + # cut at the nulls + data = data[:end_idx] + + # For utf16 and utf32, just looking for the nulls cuts the final null from the string when its ascii characters + # This occurs as the string 'vol.py' in utf-16 will look like this, with two ending nulls: + # "v\x00o\x00l\x00.\x00p\x00y\x00\x00\x00" + # By cutting at the first \x00\x00, we are taking the second byte of the character for 'y' + # With real unicode strings this character can be non-zero + # This check and added null, pads out the last byte(s) to the width of each character to avoid this issue + end_size = len(ending_nulls) + if len(data) > end_size and len(data) % end_size != 0: + data += b"\x00" * (end_size - (len(data) % end_size)) + + return data.decode(encoding=encoding, errors=errors) def array_of_pointers(