Properly reconstruct strings from memory buffers and allow for plugin-specified encodings

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 = ""
 

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(