Merge pull request #1098 from volatilityfoundation/feature/output-filter

Implement CLI support for filtering out rows during output

volatility3/cli/__init__.py

@@ -22,6 +22,7 @@ import traceback
 from typing import Any, Dict, List, Tuple, Type, Union
 from urllib import parse, request
 
+from volatility3.cli import text_filter
 import volatility3.plugins
 import volatility3.symbols
 from volatility3 import framework
@@ -233,6 +234,12 @@ class CommandLine:
             default=False,
             action="store_true",
         )
+        parser.add_argument(
+            "--filters",
+            help="List of filters to apply to the output (in the form of [+-]columname,pattern[!])",
+            default=[],
+            action="append",
+        )
 
         parser.set_defaults(**default_config)
 
@@ -454,7 +461,10 @@ class CommandLine:
         try:
             # Construct and run the plugin
             if constructed:
-                renderers[args.renderer]().render(constructed.run())
+                grid = constructed.run()
+                renderer = renderers[args.renderer]()
+                renderer.filter = text_filter.CLIFilter(grid, args.filters)
+                renderer.render(grid)
         except exceptions.VolatilityException as excp:
             self.process_exceptions(excp)
 

volatility3/cli/text_filter.py

@@ -0,0 +1,98 @@
+import logging
+from typing import Any, List, Optional
+from volatility3.framework import constants, interfaces
+import re
+
+vollog = logging.getLogger(__name__)
+
+
+class CLIFilter:
+    def __init__(self, treegrid, filters: List[str]):
+        self._filters = self._prepare(treegrid, filters)
+
+    def _prepare(self, treegrid: interfaces.renderers.TreeGrid, filters: List[str]):
+        """Runs through the filter strings and creates the necessary filter objects"""
+        output = []
+
+        for filter in filters:
+            exclude = False
+            regex = False
+            pattern = None
+            column_name = None
+            if filter.startswith("-"):
+                exclude = True
+                filter = filter[1:]
+            elif filter.startswith("+"):
+                filter = filter[1:]
+            components = filter.split(",")
+            if len(components) < 2:
+                pattern = components[0]
+            else:
+                column_name = components[0]
+                pattern = ",".join(components[1:])
+            if pattern and pattern.endswith("!"):
+                regex = True
+                pattern = pattern[:-1]
+            column_num = None
+            if column_name:
+                for num, column in enumerate(treegrid.columns):
+                    if column_name.lower() in column.name.lower():
+                        column_num = num
+                        break
+            if pattern:
+                output.append(ColumnFilter(column_num, pattern, regex, exclude))
+
+        vollog.log(constants.LOGLEVEL_VVV, "Filters:\n" + repr(output))
+
+        return output
+
+    def filter(
+        self,
+        row: List[Any],
+    ) -> bool:
+        """Filters the row based on each of the column_filters"""
+        if not self._filters:
+            return False
+        found = any([column_filter.found(row) for column_filter in self._filters])
+        return not found
+
+
+class ColumnFilter:
+    def __init__(
+        self,
+        column_num: Optional[int],
+        pattern: str,
+        regex: bool = False,
+        exclude: bool = False,
+    ) -> None:
+        self.column_num = column_num
+        self.pattern = pattern
+        self.exclude = exclude
+        self.regex = regex
+
+    def find(self, item) -> bool:
+        """Identifies whether an item is found in the appropriate column"""
+        try:
+            if self.regex:
+                return re.search(self.pattern, f"{item}")
+            return self.pattern in f"{item}"
+        except IOError:
+            return False
+
+    def found(self, row: List[Any]) -> bool:
+        """Determines whether a row should be filtered
+
+        If the classes exclude value is false, and the necessary pattern is found, the row is not filtered,
+        otherwise it is filtered.
+        """
+        if self.column_num is None:
+            found = any([self.find(x) for x in row])
+        else:
+            found = self.find(row[self.column_num])
+        if self.exclude:
+            return not found
+        return found
+
+    def __repr__(self) -> str:
+        """Returns a display of a column filter"""
+        return f"ColumnFilter(column={self.column_num},exclude={self.exclude},regex={self.regex},pattern={self.pattern})"

volatility3/cli/text_renderer.py

@@ -10,6 +10,7 @@ import string
 import sys
 from functools import wraps
 from typing import Any, Callable, Dict, List, Tuple
+from volatility3.cli import text_filter
 
 from volatility3.framework import interfaces, renderers
 from volatility3.framework.renderers import format_hints
@@ -134,6 +135,7 @@ class CLIRenderer(interfaces.renderers.Renderer):
 
     name = "unnamed"
     structured_output = False
+    filter: text_filter.CLIFilter = None
 
 
 class QuickTextRenderer(CLIRenderer):
@@ -172,6 +174,9 @@ class QuickTextRenderer(CLIRenderer):
         outfd.write("\n{}\n".format("\t".join(line)))
 
         def visitor(node: interfaces.renderers.TreeNode, accumulator):
+            if self.filter and self.filter.filter(node.values):
+                return accumulator
+
             accumulator.write("\n")
             # Nodes always have a path value, giving them a path_depth of at least 1, we use max just in case
             accumulator.write(
@@ -306,6 +311,10 @@ class PrettyTextRenderer(CLIRenderer):
             max_column_widths[tree_indent_column] = max(
                 max_column_widths.get(tree_indent_column, 0), node.path_depth
             )
+
+            if self.filter and self.filter.filter(node.values):
+                return accumulator
+
             line = {}
             for column_index in range(len(grid.columns)):
                 column = grid.columns[column_index]