Vastly improve string reading while keeping intended behaviour

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(