Merge branch 'master' of https://github.com/volatilityfoundation/volatility3

volatility/framework/plugins/windows/poolscanner.py

@@ -20,7 +20,7 @@
 
 import enum
 import logging
-from typing import Dict, Generator, List, Optional, Tuple
+from typing import Dict, Generator, List, Optional, Tuple, Callable
 
 import volatility.plugins.windows.handles as handles
 
@@ -72,6 +72,57 @@ class PoolConstraint:
         self.alignment = alignment
 
 
+def os_distinguisher(version_check: Callable[[Tuple[int, ...]], bool],
+                     fallback_checks: List[Tuple[str, Optional[str], bool]]
+                     ) -> Callable[[interfaces.context.ContextInterface, str], bool]:
+    """Distinguishes a symbol table as being above a particular version or point
+
+       This will primarily check the version metadata first and foremost.
+       If that metadata isn't available then each item in the fallback_checks is tested.
+       If invert is specified then the result will be true if the version is less than that specified, or in the case of
+       fallback, if any of the fallback checks is successful.
+
+       A fallback check is made up of:
+        * a symbol or type name
+        * a member name (implying that the value before was a type name)
+        * whether that symbol, type or member must be present or absent for the symbol table to be more above the required point
+
+        Note: Specifying that a member must not be present includes the whole type not being present too (ie, either will pass the test)
+    """
+
+    # try the primary method based on the pe version in the ISF
+    def method(context: interfaces.context.ContextInterface, symbol_table: str) -> bool:
+
+        try:
+            pe_version = context.symbol_space[symbol_table].metadata.pe_version
+            major, minor, revision, build = pe_version
+            return version_check((major, minor, revision, build))
+        except (AttributeError, ValueError, TypeError):
+            vollog.log(constants.LOGLEVEL_VVV, "Windows PE version data is not available")
+
+        if not fallback_checks:
+            raise ValueError("No fallback methods for os_distinguishing provided")
+
+        # fall back to the backup method, if necessary
+        for name, member, response in fallback_checks:
+            if member is None:
+                if (context.symbol_space.has_symbol(symbol_table + constants.BANG + name)
+                        or context.symbol_space.has_type(symbol_table + constants.BANG + name)) != response:
+                    return False
+            else:
+                try:
+                    symbol_type = context.symbol_space.get_type(symbol_table + constants.BANG + name)
+                    if symbol_type.has_member(member) != response:
+                        return False
+                except exceptions.SymbolError:
+                    if not response:
+                        return False
+
+        return True
+
+    return method
+
+
 class PoolScanner(plugins.PluginInterface):
     """A generic pool scanner plugin"""
 
@@ -83,61 +134,14 @@ class PoolScanner(plugins.PluginInterface):
             requirements.SymbolTableRequirement(name = "nt_symbols", description = "Windows kernel symbols")
         ]
 
-    @staticmethod
-    def is_windows_10(context: interfaces.context.ContextInterface, symbol_table: str) -> bool:
-        """Determine if the analyzed sample is Windows 10"""
-
-        # try the primary method based on the pe version in the ISF
-        try:
-            pe_version = context.symbol_space[symbol_table].metadata.pe_version
-            major, minor, _revision, _build = pe_version
-            return (major, minor) >= (10, 0)
-        except (AttributeError, ValueError, TypeError):
-            vollog.log(constants.LOGLEVEL_VVV, "Windows PE version data is not available")
-
-        # fall back to the backup method, if necessary
-        try:
-            _symbol = context.symbol_space.get_symbol(symbol_table + constants.BANG + "ObHeaderCookie")
-            return True
-        except exceptions.SymbolError:
-            return False
-
-    @staticmethod
-    def is_windows_8_or_later(context: interfaces.context.ContextInterface, layer_name: str, symbol_table: str) -> bool:
-        """Determine if the analyzed sample is Windows 8 or later"""
-
-        # try the primary method based on the pe version in the ISF
-        try:
-            pe_version = context.symbol_space[symbol_table].metadata.pe_version
-            major, minor, _revision, _build = pe_version
-            return (major, minor) >= (6, 2)
-        except (AttributeError, ValueError, TypeError):
-            vollog.log(constants.LOGLEVEL_VVV, "Windows PE version data is not available")
-
-        # fall back to the backup method, if necessary
-        kvo = context.memory[layer_name].config['kernel_virtual_offset']
-        ntkrnlmp = context.module(symbol_table, layer_name = layer_name, offset = kvo)
-        handle_table_type = ntkrnlmp.get_type("_HANDLE_TABLE")
-        return not handle_table_type.has_member("HandleCount")
-
-    @staticmethod
-    def is_windows_7(context: interfaces.context.ContextInterface, layer_name: str, symbol_table: str) -> bool:
-        """Determine if the analyzed sample is Windows 7"""
-
-        # try the primary method based on the pe version in the ISF
-        try:
-            pe_version = context.symbol_space[symbol_table].metadata.pe_version
-            major, minor, _revision, _build = pe_version
-            return (major, minor) == (6, 1)
-        except (AttributeError, ValueError):
-            vollog.log(constants.LOGLEVEL_VVV, "Windows PE version data is not available")
-
-        # fall back to the backup method, if necessary
-        kvo = context.memory[layer_name].config['kernel_virtual_offset']
-        ntkrnlmp = context.module(symbol_table, layer_name = layer_name, offset = kvo)
-        handle_table_type = ntkrnlmp.get_type("_OBJECT_HEADER")
-        return (handle_table_type.has_member("TypeIndex")
-                and not PoolScanner.is_windows_8_or_later(context, layer_name, symbol_table))
+    is_windows_10 = os_distinguisher(
+        version_check = lambda x: x >= (10, 0), fallback_checks = [("ObHeaderCookie", None, True)])
+    is_windows_8_or_later = os_distinguisher(
+        version_check = lambda x: x >= (6, 2), fallback_checks = [("_HANDLE_TABLE", "HandleCount", False)])
+    # Technically, this is win7 or less
+    is_windows_7 = os_distinguisher(
+        version_check = lambda x: x == (6, 1),
+        fallback_checks = [("_OBJECT_HEADER", "TypeIndex", True), ("_HANDLE_TABLE", "HandleCount", True)])
 
     def _generator(self):
 
@@ -281,8 +285,7 @@ class PoolScanner(plugins.PluginInterface):
         cookie = handles.Handles.find_cookie(context = context, layer_name = layer_name, symbol_table = symbol_table)
 
         is_windows_10 = cls.is_windows_10(context = context, symbol_table = symbol_table)
-        is_windows_8_or_later = cls.is_windows_8_or_later(
-            context = context, layer_name = layer_name, symbol_table = symbol_table)
+        is_windows_8_or_later = cls.is_windows_8_or_later(context = context, symbol_table = symbol_table)
 
         # start off with the primary virtual layer
         scan_layer = layer_name
@@ -332,7 +335,7 @@ class PoolScanner(plugins.PluginInterface):
             # We have to manually load a symbol table
 
             if symbols.symbol_table_is_64bit(context, symbol_table):
-                is_win_7 = PoolScanner.is_windows_7(context, 'primary', symbol_table)
+                is_win_7 = cls.is_windows_7(context = context, symbol_table = symbol_table)
                 if is_win_7:
                     pool_header_json_filename = "poolheader-x64-win7"
                 else: