vadinfo: Add ability to filter processes by physical memory offset and display physical VAD offsets.

volatility3/framework/plugins/windows/vadinfo.py

@@ -9,7 +9,7 @@
 from volatility3.framework.configuration import requirements
 from volatility3.framework.objects import utility
 from volatility3.framework.renderers import format_hints
-from volatility3.plugins.windows import pslist
+from volatility3.plugins.windows import pslist, psscan
 
 vollog = logging.getLogger(__name__)
 
@@ -79,6 +79,16 @@ def get_requirements(cls) -> List[interfaces.configuration.RequirementInterface]
                 default=cls.MAXSIZE_DEFAULT,
                 optional=True,
             ),
+            requirements.IntRequirement(
+                name="offset",
+                description="Process offset in the physical address space",
+                optional=True,
+            ),
+            requirements.BooleanRequirement(
+                name="physical-offsets",
+                description="List processes with physical offsets instead of virtual offsets.",
+                optional=True,
+            ),
         ]
 
     @classmethod
@@ -198,6 +208,22 @@ def vad_dump(
 
         return file_handle
 
+    def _translate_offset(self, offset: int) -> int:
+        if not self.config["physical-offsets"]:
+            return offset
+
+        kernel = self.context.modules[self.config["kernel"]]
+        layer_name = kernel.layer_name
+
+        try:
+            _original_offset, _original_length, offset, _length, _layer_name = list(
+                self.context.layers[layer_name].mapping(offset=offset, length=0)
+            )[0]
+        except exceptions.PagedInvalidAddressException:
+            vollog.debug(f"Page fault: unable to translate {offset:0x}")
+
+        return offset
+
     def _generator(self, procs: List[interfaces.objects.ObjectInterface]) -> Generator[
         Tuple[
             int,
@@ -252,7 +278,7 @@ def filter_function(x: interfaces.objects.ObjectInterface) -> bool:
                     (
                         proc.UniqueProcessId,
                         process_name,
-                        format_hints.Hex(kernel_layer.canonicalize(vad.vol.offset)),
+                        format_hints.Hex(kernel_layer.canonicalize(self._translate_offset(vad.vol.offset))),
                         format_hints.Hex(vad.get_start()),
                         format_hints.Hex(vad.get_end()),
                         vad.get_tag(),
@@ -274,6 +300,24 @@ def filter_function(x: interfaces.objects.ObjectInterface) -> bool:
 
     def run(self) -> renderers.TreeGrid:
         filter_func = pslist.PsList.create_pid_filter(self.config.get("pid", None))
+        kernel = self.context.modules[self.config["kernel"]]
+
+        if self.config["offset"]:
+            procs = psscan.PsScan.scan_processes(
+                self.context,
+                self.config["kernel"],
+                filter_func=psscan.PsScan.create_offset_filter(
+                    self.context,
+                    kernel.layer_name,
+                    self.config["offset"],
+                ),
+            )
+        else:
+             procs = pslist.PsList.list_processes(
+                context=self.context,
+                kernel_module_name=self.config["kernel"],
+                filter_func=filter_func,
+            )
 
         return renderers.TreeGrid(
             [
@@ -290,11 +334,5 @@ def run(self) -> renderers.TreeGrid:
                 ("File", str),
                 ("File output", str),
             ],
-            self._generator(
-                pslist.PsList.list_processes(
-                    context=self.context,
-                    kernel_module_name=self.config["kernel"],
-                    filter_func=filter_func,
-                )
-            ),
+            self._generator(procs=procs),
         )