Merge branch 'develop' into additional_windows_testing

Author: Abyss-W4tcher

.github/workflows/ruff.yaml

@@ -9,7 +9,7 @@ jobs:
     steps:
       - uses: actions/checkout@v4
 
-      - uses: astral-sh/ruff-action@v1
+      - uses: astral-sh/ruff-action@v3.2.1
         with:
           args: check
           src: "."

doc/source/simple-plugin.rst

@@ -53,9 +53,9 @@ to be able to run properly.  Any that are defined as optional need not necessari
                     description = "Process IDs to include (all other processes are excluded)",
                     optional = True
                 ),
-                requirements.PluginRequirement(
+                requirements.VersionRequirement(
                     name = 'pslist',
-                    plugin = pslist.PsList,
+                    component = pslist.PsList,
                     version = (2, 0, 0)
                 ),
             ]

volatility3/cli/volshell/generic.py

@@ -8,6 +8,7 @@
 import string
 import struct
 import sys
+import textwrap
 from typing import Any, Dict, Iterable, List, Optional, Tuple, Type, Union
 from urllib import parse, request
 
@@ -23,6 +24,14 @@
 except ImportError:
     has_capstone = False
 
+try:
+    from IPython import terminal
+    from traitlets import config as traitlets_config
+
+    has_ipython = True
+except ImportError:
+    has_ipython = False
+
 
 class Volshell(interfaces.plugins.PluginInterface):
     """Shell environment to directly interact with a memory image."""
@@ -51,7 +60,13 @@ def get_requirements(cls) -> List[interfaces.configuration.RequirementInterface]
                     description="File to load and execute at start",
                     default=None,
                     optional=True,
-                )
+                ),
+                requirements.BooleanRequirement(
+                    name="script-only",
+                    description="Exit volshell after the script specified in --script completes",
+                    default=False,
+                    optional=True,
+                ),
             ]
         return reqs + [
             requirements.TranslationLayerRequirement(
@@ -69,43 +84,70 @@ def run(
         """
 
         # Try to enable tab completion
-        try:
-            import readline
-        except ImportError:
-            pass
-        else:
-            import rlcompleter
+        if not has_ipython:
+            try:
+                import readline
+                import rlcompleter
 
-            completer = rlcompleter.Completer(namespace=self._construct_locals_dict())
-            readline.set_completer(completer.complete)
-            readline.parse_and_bind("tab: complete")
-            print("Readline imported successfully")
+                completer = rlcompleter.Completer(
+                    namespace=self._construct_locals_dict()
+                )
+                readline.set_completer(completer.complete)
+                readline.parse_and_bind("tab: complete")
+                print("Readline imported successfully")
+            except ImportError:
+                print(
+                    "Readline or rlcompleter module could not be imported. Tab completion will not be available."
+                )
 
         # TODO: provide help, consider generic functions (pslist?) and/or providing windows/linux functions
 
         mode = self.__module__.split(".")[-1]
         mode = mode[0].upper() + mode[1:]
 
-        banner = f"""
-    Call help() to see available functions
+        banner = textwrap.dedent(
+            f"""
+            Call help() to see available functions
 
-    Volshell mode        : {mode}
-    Current Layer        : {self.current_layer}
-    Current Symbol Table : {self.current_symbol_table}
-    Current Kernel Name  : {self.current_kernel_name}
-"""
+            Volshell mode        : {mode}
+            Current Layer        : {self.current_layer}
+            Current Symbol Table : {self.current_symbol_table}
+            Current Kernel Name  : {self.current_kernel_name}
+            """
+        )
 
         sys.ps1 = f"({self.current_layer}) >>> "
         # Dict self._construct_locals_dict() will have priority on keys
         combined_locals = additional_locals.copy()
         combined_locals.update(self._construct_locals_dict())
-        self.__console = code.InteractiveConsole(locals=combined_locals)
+        if has_ipython:
+
+            class LayerNamePrompt(terminal.prompts.Prompts):
+                def in_prompt_tokens(self, cli=None):
+                    slf = self.shell.user_ns.get("self")
+                    layer_name = slf.current_layer if slf else "no_layer"
+                    return [(terminal.prompts.Token.Prompt, f"[{layer_name}]> ")]
+
+            c = traitlets_config.Config()
+            c.TerminalInteractiveShell.prompts_class = LayerNamePrompt
+            c.InteractiveShellEmbed.banner2 = banner
+            self.__console = terminal.embed.InteractiveShellEmbed(
+                config=c, user_ns=combined_locals
+            )
+        else:
+            self.__console = code.InteractiveConsole(locals=combined_locals)
         # Since we have to do work to add the option only once for all different modes of volshell, we can't
         # rely on the default having been set
         if self.config.get("script", None) is not None:
             self.run_script(location=self.config["script"])
 
-        self.__console.interact(banner=banner)
+            if self.config.get("script-only"):
+                exit()
+
+        if has_ipython:
+            self.__console()
+        else:
+            self.__console.interact(banner=banner)
 
         return renderers.TreeGrid([("Terminating", str)], None)
 
@@ -277,23 +319,25 @@ def display_bytes(self, offset, count=DEFAULT_NUM_DISPLAY_BYTES, layer_name=None
         self._display_data(offset, remaining_data)
 
     def display_quadwords(
-        self, offset, count=DEFAULT_NUM_DISPLAY_BYTES, layer_name=None
+        self, offset, count=DEFAULT_NUM_DISPLAY_BYTES, layer_name=None, byteorder="@"
     ):
         """Displays quad-word values (8 bytes) and corresponding ASCII characters"""
         remaining_data = self._read_data(offset, count=count, layer_name=layer_name)
-        self._display_data(offset, remaining_data, format_string="Q")
+        self._display_data(offset, remaining_data, format_string=f"{byteorder}Q")
 
     def display_doublewords(
-        self, offset, count=DEFAULT_NUM_DISPLAY_BYTES, layer_name=None
+        self, offset, count=DEFAULT_NUM_DISPLAY_BYTES, layer_name=None, byteorder="@"
     ):
         """Displays double-word values (4 bytes) and corresponding ASCII characters"""
         remaining_data = self._read_data(offset, count=count, layer_name=layer_name)
-        self._display_data(offset, remaining_data, format_string="I")
+        self._display_data(offset, remaining_data, format_string=f"{byteorder}I")
 
-    def display_words(self, offset, count=DEFAULT_NUM_DISPLAY_BYTES, layer_name=None):
+    def display_words(
+        self, offset, count=DEFAULT_NUM_DISPLAY_BYTES, layer_name=None, byteorder="@"
+    ):
         """Displays word values (2 bytes) and corresponding ASCII characters"""
         remaining_data = self._read_data(offset, count=count, layer_name=layer_name)
-        self._display_data(offset, remaining_data, format_string="H")
+        self._display_data(offset, remaining_data, format_string=f"{byteorder}H")
 
     def regex_scan(self, pattern, count=DEFAULT_NUM_DISPLAY_BYTES, layer_name=None):
         """Scans for regex pattern in layer using RegExScanner."""
@@ -508,10 +552,13 @@ def run_script(self, location: str):
             location = "file:" + request.pathname2url(location)
         print(f"Running code from {location}\n")
         accessor = resources.ResourceAccessor()
-        with accessor.open(url=location) as fp:
-            self.__console.runsource(
-                io.TextIOWrapper(fp, encoding="utf-8").read(), symbol="exec"
-            )
+        with accessor.open(url=location) as handle, io.TextIOWrapper(
+            handle, encoding="utf-8"
+        ) as fp:
+            if has_ipython:
+                self.__console.ex(fp.read())
+            else:
+                self.__console.runsource(fp.read(), symbol="exec")
         print("\nCode complete")
 
     def load_file(self, location: str):

volatility3/cli/volshell/linux.py

@@ -30,8 +30,8 @@ def get_requirements(cls):
             requirements.ModuleRequirement(
                 name="kernel", description="Linux kernel module"
             ),
-            requirements.PluginRequirement(
-                name="pslist", plugin=pslist.PsList, version=(4, 0, 0)
+            requirements.VersionRequirement(
+                name="pslist", component=pslist.PsList, version=(4, 0, 0)
             ),
             requirements.IntRequirement(
                 name="pid", description="Process ID", optional=True

volatility3/cli/volshell/mac.py

@@ -19,8 +19,8 @@ def get_requirements(cls):
             requirements.ModuleRequirement(
                 name="kernel", description="Darwin kernel module"
             ),
-            requirements.PluginRequirement(
-                name="pslist", plugin=pslist.PsList, version=(3, 0, 0)
+            requirements.VersionRequirement(
+                name="pslist", component=pslist.PsList, version=(3, 0, 0)
             ),
             requirements.IntRequirement(
                 name="pid", description="Process ID", optional=True

volatility3/cli/volshell/windows.py

@@ -17,8 +17,8 @@ class Volshell(generic.Volshell):
     def get_requirements(cls):
         return [
             requirements.ModuleRequirement(name="kernel", description="Windows kernel"),
-            requirements.PluginRequirement(
-                name="pslist", plugin=pslist.PsList, version=(3, 0, 0)
+            requirements.VersionRequirement(
+                name="pslist", component=pslist.PsList, version=(3, 0, 0)
             ),
             requirements.IntRequirement(
                 name="pid", description="Process ID", optional=True

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 = 25  # 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/__init__.py

@@ -402,13 +402,35 @@ def _unmarshall(
         pointer should be recast.  The "pointer" must always live within
         the space (even if the data provided is invalid).
         """
+        mask = context.layers[object_info.native_layer_name].address_mask
+        new = (
+            cls._get_raw_value(
+                context, data_format, object_info.layer_name, object_info.offset
+            )
+            & mask
+        )
+        return new
+
+    @classmethod
+    def _get_raw_value(
+        cls,
+        context: interfaces.context.ContextInterface,
+        data_format: DataFormatInfo,
+        layer_name: str,
+        offset: int,
+    ) -> int:
         length, endian, signed = data_format
         if signed:
             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)
+        data = context.layers.read(layer_name, offset, length)
         value = int.from_bytes(data, byteorder=endian, signed=signed)
-        return value & mask
+        return value
+
+    def get_raw_value(self) -> int:
+        raw = self._get_raw_value(
+            self._context, self.vol.data_format, self.vol.layer_name, self.vol.offset
+        )
+        return raw
 
     def dereference(
         self, layer_name: Optional[str] = None