From 48a5736a576a5b9c6119d64d85a46c8cc370bc31 Mon Sep 17 00:00:00 2001 From: Andrew Case Date: Thu, 13 Mar 2025 21:09:38 +0000 Subject: [PATCH 1/7] Properly reconstruct strings from memory buffers and allow for plugin-specified encodings --- volatility3/framework/constants/_version.py | 2 +- volatility3/framework/objects/utility.py | 27 ++++++++++++--------- 2 files changed, 17 insertions(+), 12 deletions(-) diff --git a/volatility3/framework/constants/_version.py b/volatility3/framework/constants/_version.py index aa8e8936f..1ea59c068 100644 --- a/volatility3/framework/constants/_version.py +++ b/volatility3/framework/constants/_version.py @@ -1,6 +1,6 @@ # We use the SemVer 2.0.0 versioning scheme VERSION_MAJOR = 2 # Number of releases of the library with a breaking change -VERSION_MINOR = 23 # Number of changes that only add to the interface +VERSION_MINOR = 24 # Number of changes that only add to the interface VERSION_PATCH = 0 # Number of changes that do not change the interface VERSION_SUFFIX = "" diff --git a/volatility3/framework/objects/utility.py b/volatility3/framework/objects/utility.py index 500c0e9a5..1fcd305f7 100644 --- a/volatility3/framework/objects/utility.py +++ b/volatility3/framework/objects/utility.py @@ -2,6 +2,7 @@ # 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 @@ -33,6 +34,7 @@ def array_to_string( count: Optional[int] = None, errors: str = "replace", block_size=32, + encoding="utf-8", ) -> str: """Takes a Volatility 'Array' of characters and returns a Python string. @@ -60,6 +62,7 @@ def array_to_string( count=count, errors=errors, block_size=block_size, + encoding=encoding, ) @@ -68,6 +71,7 @@ def pointer_to_string( count: int, errors: str = "replace", block_size=32, + encoding="utf-8", ) -> str: """Takes a Volatility 'Pointer' to characters and returns a Python string. @@ -94,6 +98,7 @@ def pointer_to_string( count=count, errors=errors, block_size=block_size, + encoding=encoding, ) @@ -104,6 +109,7 @@ def address_to_string( count: int, errors: str = "replace", block_size=32, + encoding="utf-8", ) -> str: """Reads a null-terminated string from a given specified memory address, processing it in blocks for efficiency. @@ -126,18 +132,17 @@ def address_to_string( raise ValueError("Count must be greater than 0") layer = context.layers[layer_name] - text = b"" - while len(text) < count: - current_block_size = min(count - len(text), block_size) - temp_text = layer.read(address + len(text), current_block_size) - idx = temp_text.find(b"\x00") - if idx != -1: - temp_text = temp_text[:idx] - text += temp_text - break - text += temp_text - return text.decode(errors=errors) + # Purposely do not catch exception + data = layer.read(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) + + return decoded_data[:idx] def array_of_pointers( From 7ab62365fbbaf3184942eed0a13848cd8781ef00 Mon Sep 17 00:00:00 2001 From: Andrew Case Date: Fri, 14 Mar 2025 03:05:11 +0000 Subject: [PATCH 2/7] Vastly improve string reading while keeping intended behaviour --- volatility3/framework/objects/utility.py | 97 +++++++++++++++++++++--- 1 file changed, 87 insertions(+), 10 deletions(-) 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( From 36b3c2885974e7c697a160bfc8820800e680286b Mon Sep 17 00:00:00 2001 From: Andrew Case Date: Fri, 14 Mar 2025 03:12:06 +0000 Subject: [PATCH 3/7] Update encodings --- volatility3/framework/objects/utility.py | 9 ++++++++- 1 file changed, 8 insertions(+), 1 deletion(-) diff --git a/volatility3/framework/objects/utility.py b/volatility3/framework/objects/utility.py index a1ad4fdf5..13031e1f1 100644 --- a/volatility3/framework/objects/utility.py +++ b/volatility3/framework/objects/utility.py @@ -188,7 +188,14 @@ def address_to_string( if count < 1: raise ValueError("Count must be greater than 0") - encodings = {"utf8": 1, "utf16": 2, "utf32": 4} + encodings = { + "utf-8": 1, + "utf8": 1, + "utf-16": 2, + "utf16": 2, + "utf32": 4, + "utf-32": 4, + } if encoding not in encodings: raise ValueError( f"Encoding ({encoding} is invalid. Must be one of {[e for e in encodings]}." From 4225adce56d1c73ff0194d944b948ad6befa2b87 Mon Sep 17 00:00:00 2001 From: Andrew Case Date: Fri, 14 Mar 2025 16:32:54 +0000 Subject: [PATCH 4/7] Convert to .mapping and let Python had all encodings --- volatility3/framework/objects/utility.py | 131 ++++++++++------------- 1 file changed, 58 insertions(+), 73 deletions(-) diff --git a/volatility3/framework/objects/utility.py b/volatility3/framework/objects/utility.py index 13031e1f1..b014e37fa 100644 --- a/volatility3/framework/objects/utility.py +++ b/volatility3/framework/objects/utility.py @@ -2,6 +2,8 @@ # 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, exceptions @@ -101,7 +103,9 @@ def pointer_to_string( ) -def gather_contiguous_bytes_from_address(layer, address: int, count: int) -> bytes: +def gather_contiguous_bytes_from_address( + context, data_layer, starting_address: int, count: int +) -> bytes: """ This method reconstructs a string from memory while also carefully examining each page @@ -115,50 +119,65 @@ def gather_contiguous_bytes_from_address(layer, address: int, count: int) -> byt data = b"" - left_to_read = count + last_address = None - # 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 + for address, length, _, _, _ in data_layer.mapping( + offset=starting_address, length=count, ignore_errors=True + ): + # Used to track when we hit a paged out page + if not last_address: + last_address = address + length - # 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 + # we hit a swapped out page + elif last_address and last_address != address: + break - left_to_read = 0 + data += data_layer.read(address, length) - 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 + # if we were able to read from the first page, we want to try and construct the string + # if the first page fails -> throw exception + if data: + return data + else: + raise exceptions.InvalidAddressException( + layer_name=data_layer, invalid_address=starting_address + ) return data +def bytes_to_decoded_string(data: bytes, encoding: str, errors: str) -> bytes: + """ + This function takes a bytes buffer that contains at a string of unknown + length starting at the first byte, and returns the properly decoded string + + It starts by using Python's `bytes.decode` to attempt to decode the entire string + It then finds the termination character (\ufffd or \x00) and splices the string + Finally, it returns this spliced string after its been decoded with the + caller-specified encoding + """ + # this is the standard byte used to replace bad unicode characters + unicode_replacement_char = "\ufffd" + + # used to find the terminating byte + termination_re = re.compile(f"{unicode_replacement_char}|\x00") + + # run over the entire string, letting Python replace invalid characters + full_decoded_string = data.decode(encoding=encoding, errors="replace") + + # stop at the first terminating character or get the whole string if not found + try: + idx = termination_re.search(full_decoded_string).start() + except AttributeError: + idx = len(full_decoded_string) + + # cut at terminating byte, if found + data = data[:idx] + + # return with caller-specified encoding and errors + return data.decode(encoding=encoding, errors=errors) + + def address_to_string( context: interfaces.context.ContextInterface, layer_name: str, @@ -188,45 +207,11 @@ def address_to_string( if count < 1: raise ValueError("Count must be greater than 0") - encodings = { - "utf-8": 1, - "utf8": 1, - "utf-16": 2, - "utf16": 2, - "utf32": 4, - "utf-32": 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] - data = gather_contiguous_bytes_from_address(layer, address, count) + data = gather_contiguous_bytes_from_address(context, layer, address, count) - # we need to find the ending nulls, which the amount of nulls varies based on encoding - ending_nulls = b"\x00" * encodings[encoding] - - 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) + return bytes_to_decoded_string(data=data, errors=errors, encoding=encoding) def array_of_pointers( From f38ddaf7155dce6c9171ad052bbfb6db0c9be3a8 Mon Sep 17 00:00:00 2001 From: Andrew Case Date: Fri, 14 Mar 2025 20:53:02 +0000 Subject: [PATCH 5/7] Fix string scanning code --- volatility3/framework/objects/utility.py | 30 ++++++++++++++---------- 1 file changed, 17 insertions(+), 13 deletions(-) diff --git a/volatility3/framework/objects/utility.py b/volatility3/framework/objects/utility.py index b014e37fa..f1ee701bf 100644 --- a/volatility3/framework/objects/utility.py +++ b/volatility3/framework/objects/utility.py @@ -119,20 +119,26 @@ def gather_contiguous_bytes_from_address( data = b"" - last_address = None + if isinstance(data_layer, interfaces.layers.TranslationLayerInterface): + last_address = None + + for address, length, _, _, _ in data_layer.mapping( + offset=starting_address, length=count, ignore_errors=True + ): + # Used to track when we hit a paged out page + if not last_address: + last_address = address + length + + # we hit a swapped out page + elif last_address and last_address != address: + break + + data += data_layer.read(address, length) - for address, length, _, _, _ in data_layer.mapping( - offset=starting_address, length=count, ignore_errors=True - ): - # Used to track when we hit a paged out page - if not last_address: last_address = address + length - # we hit a swapped out page - elif last_address and last_address != address: - break - - data += data_layer.read(address, length) + elif starting_address + count < data_layer.maximum_address: + data = data_layer.read(starting_address, count) # if we were able to read from the first page, we want to try and construct the string # if the first page fails -> throw exception @@ -143,8 +149,6 @@ def gather_contiguous_bytes_from_address( layer_name=data_layer, invalid_address=starting_address ) - return data - def bytes_to_decoded_string(data: bytes, encoding: str, errors: str) -> bytes: """ From bb0a17004f004b6eac4113a3e9377acdc4bc6e6c Mon Sep 17 00:00:00 2001 From: Andrew Case Date: Fri, 14 Mar 2025 20:59:15 +0000 Subject: [PATCH 6/7] Add return_truncated for plugin-specified handling of truncated strings --- volatility3/framework/objects/utility.py | 20 ++++++++++++++++++-- 1 file changed, 18 insertions(+), 2 deletions(-) diff --git a/volatility3/framework/objects/utility.py b/volatility3/framework/objects/utility.py index f1ee701bf..a29b65875 100644 --- a/volatility3/framework/objects/utility.py +++ b/volatility3/framework/objects/utility.py @@ -150,8 +150,19 @@ def gather_contiguous_bytes_from_address( ) -def bytes_to_decoded_string(data: bytes, encoding: str, errors: str) -> bytes: +def bytes_to_decoded_string( + data: bytes, encoding: str, errors: str, return_truncated: bool = True +) -> bytes: """ + Args: + data: The `bytes` buffer containing the string of a string at offset 0 + encoding: An encoding value for the encoding paramater of `bytes.decode` + errors: An errors value for the errors parameter of `bytes.decode` + return_truncated: Dictates whether truncated strings should be returned or + if a ValueError should be thrown if a truncated (broken) string was decoded + Returns: + bytes: The decoded string starting at offset of data + This function takes a bytes buffer that contains at a string of unknown length starting at the first byte, and returns the properly decoded string @@ -173,7 +184,12 @@ def bytes_to_decoded_string(data: bytes, encoding: str, errors: str) -> bytes: try: idx = termination_re.search(full_decoded_string).start() except AttributeError: - idx = len(full_decoded_string) + if return_truncated: + idx = len(full_decoded_string) + else: + raise ValueError( + "return_truncated set to False and truncated string decoded." + ) # cut at terminating byte, if found data = data[:idx] From b2ec21fcf28e657f42f178c27c144af7d1d7893e Mon Sep 17 00:00:00 2001 From: Andrew Case Date: Fri, 14 Mar 2025 21:15:03 +0000 Subject: [PATCH 7/7] Simplify last_address handling --- volatility3/framework/objects/utility.py | 8 ++------ 1 file changed, 2 insertions(+), 6 deletions(-) diff --git a/volatility3/framework/objects/utility.py b/volatility3/framework/objects/utility.py index a29b65875..018b4a138 100644 --- a/volatility3/framework/objects/utility.py +++ b/volatility3/framework/objects/utility.py @@ -120,17 +120,13 @@ def gather_contiguous_bytes_from_address( data = b"" if isinstance(data_layer, interfaces.layers.TranslationLayerInterface): - last_address = None + last_address = starting_address for address, length, _, _, _ in data_layer.mapping( offset=starting_address, length=count, ignore_errors=True ): - # Used to track when we hit a paged out page - if not last_address: - last_address = address + length - # we hit a swapped out page - elif last_address and last_address != address: + if last_address != address: break data += data_layer.read(address, length)