From 207941759dc47d14fc020701fae7053dbda78dac Mon Sep 17 00:00:00 2001 From: SolitudePy <47316655+SolitudePy@users.noreply.github.com> Date: Sun, 1 Jun 2025 19:38:04 +0300 Subject: [PATCH 01/14] pebmasquerade plugin --- .../plugins/windows/pebmasquerade.py | 328 ++++++++++++++++++ 1 file changed, 328 insertions(+) create mode 100644 volatility3/framework/plugins/windows/pebmasquerade.py diff --git a/volatility3/framework/plugins/windows/pebmasquerade.py b/volatility3/framework/plugins/windows/pebmasquerade.py new file mode 100644 index 000000000..dc5293271 --- /dev/null +++ b/volatility3/framework/plugins/windows/pebmasquerade.py @@ -0,0 +1,328 @@ +import logging +import re +from pathlib import PureWindowsPath +from typing import List, Union, Tuple + +from volatility3.framework import interfaces, renderers, exceptions +from volatility3.framework.configuration import requirements +from volatility3.plugins.windows import pslist + +vollog = logging.getLogger(__name__) + + +# https://www.ired.team/offensive-security/defense-evasion/masquerading-processes-in-userland-through-_peb +# https://github.com/FuzzySecurity/PowerShell-Suite/blob/master/Masquerade-PEB.ps1 +class PebMasquerade(interfaces.plugins.PluginInterface): + """Detects potential process name spoofing by comparing EPROCESS and PEB data.""" + + _version = (1, 0, 0) + _required_framework_version = (2, 0, 0) + + @classmethod + def get_requirements(cls) -> List[interfaces.configuration.RequirementInterface]: + return [ + requirements.ModuleRequirement( + name="kernel", + description="Windows kernel", + architectures=["Intel32", "Intel64"], + ), + requirements.VersionRequirement( + name="pslist", component=pslist.PsList, version=(3, 0, 0) + ), + requirements.ListRequirement( + name="pid", + element_type=int, + description="Process ID to include (all other processes are excluded)", + optional=True, + ), + ] + + @staticmethod + def _get_cmdline_image(cmdline: str) -> Union[str, PureWindowsPath]: + """Extract the executable path from a command line string. + + Args: + cmdline (str): The command line string to parse. + + Returns: + Union[str, PureWindowsPath]: The executable path as a string or PureWindowsPath. + """ + if not cmdline: + return "" + + # Regex to extract first .exe ending string (handles quotes, paths, no quotes) + match = re.search(r'(?i)(["\']?)([^"\']*?\.exe)\1(?=\s|$)', cmdline) + if match: + exe_path = match.group(2) + return PureWindowsPath(exe_path) + + # If no .exe found, extract the first token (handles quotes) + # Matches either "quoted string" or unquoted word + first_token_match = re.match(r'\s*(?:"([^"]+)"|\'([^\']+)\'|(\S+))', cmdline) + if first_token_match: + # Extract whichever group matched + executable = ( + first_token_match.group(1) + or first_token_match.group(2) + or first_token_match.group(3) + ) + return PureWindowsPath(executable).name + ".exe" + + return "" + + @staticmethod + def _are_paths_equal(device_path: str, drive_path: str) -> Tuple[bool, str, str]: + """Compare two paths to see if they are equal, ignoring drive/device root and case. + + Args: + device_path (str): The device path (e.g. "\\Device\\HarddiskVolume1\\path") + drive_path (str): The drive path (e.g. "C:\\path") + + Returns: + tuple: (are_equal, device_path_without_drive, drive_path_without_drive) + - are_equal (bool): True if paths are equal, False otherwise + - device_path_without_drive (str): Device path without drive letter + - drive_path_without_drive (str): Drive path without drive letter + """ + pure_device_path = PureWindowsPath(device_path) + pure_drive_path = PureWindowsPath(drive_path) + device_parts = list(pure_device_path.parts) + drive_parts = list(pure_drive_path.parts) + + if pure_drive_path.is_absolute(): + new_drive_path = "/".join(drive_parts[1:]).lower() + new_device_path = "/".join(device_parts[3:]).lower() + else: + new_drive_path = "/".join(drive_parts[2:]).lower() + new_device_path = "/".join(device_parts[4:]).lower() + + return ( + new_drive_path == new_device_path, + new_device_path, + new_drive_path, + ) + + def get_process_names(self, proc: interfaces.objects.ObjectInterface) -> Tuple[ + Union[str, renderers.NotAvailableValue], + Union[str, renderers.NotAvailableValue], + Union[str, renderers.NotAvailableValue], + Union[str, renderers.NotAvailableValue], + ]: + """Extract process names and related information from various sources (EPROCESS and PEB). + + Args: + proc: The process object + + Returns: + tuple: (eprocess_imagefilename, eprocess_seaudit_imagefilename, peb_imagefilepath, peb_cmdline) + """ + eprocess_imagefilename = renderers.NotAvailableValue() + eprocess_seaudit_imagefilename = renderers.NotAvailableValue() + peb_imagefilepath = renderers.NotAvailableValue() + peb_cmdline = renderers.NotAvailableValue() + + try: + eprocess_imagefilename = proc.ImageFileName.cast( + "string", + max_length=proc.ImageFileName.vol.count, + errors="replace", + ) + except (AttributeError, exceptions.InvalidAddressException): + vollog.debug( + "Unable to read EPROCESS.ImageFileName for PID %d", proc.UniqueProcessId + ) + except Exception as e: + vollog.warning( + "Error reading EPROCESS.ImageFileName for PID %d: %s", + proc.UniqueProcessId, + str(e)[:50], + ) + + try: + audit = proc.SeAuditProcessCreationInfo.ImageFileName.Name + audit_string = audit.get_string() + if audit_string: + eprocess_seaudit_imagefilename = audit_string + except exceptions.InvalidAddressException: + vollog.debug( + "Unable to read SeAuditProcessCreationInfo.ImageFileName for PID %d", + proc.UniqueProcessId, + ) + except AttributeError: + vollog.debug( + "SeAuditProcessCreationInfo structure not available for PID %d", + proc.UniqueProcessId, + ) + except Exception as e: + vollog.warning( + "Error reading SeAuditProcessCreationInfo for PID %d: %s", + proc.UniqueProcessId, + str(e)[:50], + ) + + try: + peb = proc.get_peb() + if peb and peb.ProcessParameters: + # Get ImagePathName + try: + image_path_str = peb.ProcessParameters.ImagePathName.get_string() + if image_path_str: + peb_imagefilepath = image_path_str + except (AttributeError, exceptions.InvalidAddressException): + vollog.debug( + "Unable to read PEB.ImagePathName for PID %d", + proc.UniqueProcessId, + ) + except Exception as e: + vollog.warning( + "Error reading PEB.ImagePathName for PID %d: %s", + proc.UniqueProcessId, + str(e)[:50], + ) + + try: + cmdline_str = peb.ProcessParameters.CommandLine.get_string() + if cmdline_str: + peb_cmdline = cmdline_str + except (AttributeError, exceptions.InvalidAddressException): + vollog.debug( + "Unable to read PEB.ProcessParameters.CommandLine for PID %d", + proc.UniqueProcessId, + ) + except Exception as e: + vollog.warning( + "Error reading PEB.ProcessParameters.CommandLine for PID %d: %s", + proc.UniqueProcessId, + str(e)[:50], + ) + except (AttributeError, exceptions.InvalidAddressException): + # Important for cases where PEB does not exist or is inaccessible (e.g SYSTEM process) + vollog.debug("Unable to access PEB for PID %d", proc.UniqueProcessId) + except Exception as e: + vollog.warning( + "Error accessing PEB for PID %d: %s", proc.UniqueProcessId, str(e)[:50] + ) + + return ( + eprocess_imagefilename, + eprocess_seaudit_imagefilename, + peb_imagefilepath, + peb_cmdline, + ) + + def _generator(self): + pid_filter = pslist.PsList.create_pid_filter(self.config.get("pid", None)) + + for proc in pslist.PsList.list_processes( + context=self.context, + kernel_module_name=self.config["kernel"], + filter_func=pid_filter, + ): + proc_id = proc.UniqueProcessId + notes = [] + + ( + eprocess_imagefilename, + eprocess_seaudit_imagefilename, + peb_imagefilepath, + peb_cmdline, + ) = self.get_process_names(proc) + proc_name_for_row = eprocess_imagefilename + + # Extract command line executable path for rendering + peb_cmdline_path_render = renderers.NotAvailableValue() + if isinstance(peb_cmdline, str): + try: + peb_cmdline_path_render = str( + PebMasquerade._get_cmdline_image(peb_cmdline) + ) + except Exception as e: + vollog.debug( + "Error extracting command line path for PID %d: %s", + proc_id, + str(e)[:50], + ) + + # Populate notes for enrichment + if isinstance(eprocess_imagefilename, str) and isinstance( + peb_imagefilepath, str + ): + try: + peb_imagefilepath_basename = PureWindowsPath(peb_imagefilepath).name + peb_imagefilepath_truncated = peb_imagefilepath_basename[:14] + + # Compare EPROCESS.ImageFileName with PEB.ImageFilePath truncated to 15 characters + if ( + eprocess_imagefilename.lower() + != peb_imagefilepath_truncated.lower() + ): + notes.append( + f"'Potential PEB.ImageFilePath Spoofing: EPROCESS={eprocess_imagefilename};PEB={peb_imagefilepath_truncated}'" + ) + except Exception as e: + notes.append(f"ImageFilePath Comparison error: {str(e)[:30]}") + + if isinstance(eprocess_imagefilename, str) and isinstance(peb_cmdline, str): + try: + # Compare EPROCESS.ImageFileName with PEB.CommandLine executable path truncated to 15 characters + peb_cmdline_path = PebMasquerade._get_cmdline_image(peb_cmdline) + if isinstance(peb_cmdline_path, PureWindowsPath): + peb_cmdline_path = peb_cmdline_path.name + peb_cmdline_basename_truncated = peb_cmdline_path[:14] + if ( + eprocess_imagefilename.lower() + != peb_cmdline_basename_truncated.lower() + ): + notes.append( + f"'Potential PEB.CommandLine Spoofing: EPROCESS={eprocess_imagefilename};PEB={peb_cmdline_basename_truncated}'" + ) + except Exception as e: + notes.append(f"CommandLine comparison error: {str(e)}") + + if isinstance(eprocess_seaudit_imagefilename, str) and isinstance( + peb_imagefilepath, str + ): + try: + ( + are_equal, + eprocess_seaudit_normalized, + peb_imagefilepath_normalized, + ) = PebMasquerade._are_paths_equal( + device_path=eprocess_seaudit_imagefilename, + drive_path=peb_imagefilepath, + ) + if not are_equal: + notes.append( + f"'Potential PEB.ImageFilePath Spoofing (via _EPROCESS.SeAuditProcessCreationInfo): EPROCESS={eprocess_seaudit_normalized};PEB={peb_imagefilepath_normalized}'" + ) + except Exception as e: + notes.append( + f"SeAuditProcessCreationInfo comparison error: {str(e)[:30]}" + ) + + yield ( + 0, + ( + proc_id, + proc_name_for_row, + eprocess_imagefilename, + eprocess_seaudit_imagefilename, + peb_imagefilepath, + peb_cmdline_path_render, + "[" + ", ".join(notes) + "]" if notes else "OK", + ), + ) + + def run(self): + return renderers.TreeGrid( + [ + ("PID", int), + ("ProcessName", str), + ("EPROCESS_ImageFileName", str), + ("EPROCESS_SeAudit_ImageFileName", str), + ("PEB_ImageFilePath", str), + ("PEB_CommandLine_Path", str), + ("Notes", str), + ], + self._generator(), + ) From 72f0bd1bbafe69cc6695985a13aded63d7edc576 Mon Sep 17 00:00:00 2001 From: SolitudePy <47316655+SolitudePy@users.noreply.github.com> Date: Fri, 6 Jun 2025 17:22:39 +0300 Subject: [PATCH 02/14] UNICODE_STRING length checks to further detect spoofing --- .../plugins/windows/pebmasquerade.py | 46 +++++++++++++++++++ 1 file changed, 46 insertions(+) diff --git a/volatility3/framework/plugins/windows/pebmasquerade.py b/volatility3/framework/plugins/windows/pebmasquerade.py index dc5293271..c1bfca67f 100644 --- a/volatility3/framework/plugins/windows/pebmasquerade.py +++ b/volatility3/framework/plugins/windows/pebmasquerade.py @@ -219,6 +219,12 @@ class PebMasquerade(interfaces.plugins.PluginInterface): filter_func=pid_filter, ): proc_id = proc.UniqueProcessId + try: + peb = proc.get_peb() + except (exceptions.InvalidAddressException, AttributeError): + vollog.debug( + "Unable to access PEB for PID %d, skipping process", proc_id + ) notes = [] ( @@ -300,6 +306,46 @@ class PebMasquerade(interfaces.plugins.PluginInterface): f"SeAuditProcessCreationInfo comparison error: {str(e)[:30]}" ) + if isinstance(peb_imagefilepath, str) and peb: + try: + + # Length values are of type USHORT + peb_imagefilepath_length = ( + peb.ProcessParameters.ImagePathName.Length // 2 + ) + peb_imagefilepath_maxlength = ( + peb.ProcessParameters.ImagePathName.MaximumLength // 2 - 1 + ) + + if (peb_imagefilepath_length != len(peb_imagefilepath)) or ( + peb_imagefilepath_maxlength != len(peb_imagefilepath) + ): + notes.append( + f"'PEB.ImageFilePath Length Mismatch: Length={peb_imagefilepath_length}, MaximumLength={peb_imagefilepath_maxlength}, Actual={len(peb_imagefilepath)}'" + ) + except Exception as e: + notes.append( + f"PEB.ImageFilePath Length comparison error: {str(e)[:30]}" + ) + + if isinstance(peb_cmdline, str) and peb: + try: + # Length values are of type USHORT + peb_cmdline_length = peb.ProcessParameters.CommandLine.Length // 2 + peb_cmdline_maxlength = ( + peb.ProcessParameters.CommandLine.MaximumLength // 2 - 1 + ) + + if (peb_cmdline_length != len(peb_cmdline)) or ( + peb_cmdline_maxlength != len(peb_cmdline) + ): + notes.append( + f"'PEB.CommandLine Length Mismatch: Commandline={peb_cmdline}, Length={peb_cmdline_length}, MaximumLength={peb_cmdline_maxlength}, Actual={len(peb_cmdline)}'" + ) + except Exception as e: + notes.append( + f"PEB.CommandLine Length comparison error: {str(e)[:30]}" + ) yield ( 0, ( From 231033428df243efe16a60cd2daafef02c4ccfc7 Mon Sep 17 00:00:00 2001 From: SolitudePy <47316655+SolitudePy@users.noreply.github.com> Date: Fri, 6 Jun 2025 18:26:47 +0300 Subject: [PATCH 03/14] replace to utility function --- volatility3/framework/plugins/windows/pebmasquerade.py | 7 ++----- 1 file changed, 2 insertions(+), 5 deletions(-) diff --git a/volatility3/framework/plugins/windows/pebmasquerade.py b/volatility3/framework/plugins/windows/pebmasquerade.py index c1bfca67f..78676bc0d 100644 --- a/volatility3/framework/plugins/windows/pebmasquerade.py +++ b/volatility3/framework/plugins/windows/pebmasquerade.py @@ -5,6 +5,7 @@ from typing import List, Union, Tuple from volatility3.framework import interfaces, renderers, exceptions from volatility3.framework.configuration import requirements +from volatility3.framework.objects import utility from volatility3.plugins.windows import pslist vollog = logging.getLogger(__name__) @@ -122,11 +123,7 @@ class PebMasquerade(interfaces.plugins.PluginInterface): peb_cmdline = renderers.NotAvailableValue() try: - eprocess_imagefilename = proc.ImageFileName.cast( - "string", - max_length=proc.ImageFileName.vol.count, - errors="replace", - ) + eprocess_imagefilename = utility.array_to_string(proc.ImageFileName) except (AttributeError, exceptions.InvalidAddressException): vollog.debug( "Unable to read EPROCESS.ImageFileName for PID %d", proc.UniqueProcessId From 22ea3d55428cfde26a8ae7c4307b597696e9a9ce Mon Sep 17 00:00:00 2001 From: SolitudePy <47316655+SolitudePy@users.noreply.github.com> Date: Sat, 7 Jun 2025 13:35:27 +0300 Subject: [PATCH 04/14] moved to malware category --- .../framework/plugins/windows/{ => malware}/pebmasquerade.py | 0 1 file changed, 0 insertions(+), 0 deletions(-) rename volatility3/framework/plugins/windows/{ => malware}/pebmasquerade.py (100%) diff --git a/volatility3/framework/plugins/windows/pebmasquerade.py b/volatility3/framework/plugins/windows/malware/pebmasquerade.py similarity index 100% rename from volatility3/framework/plugins/windows/pebmasquerade.py rename to volatility3/framework/plugins/windows/malware/pebmasquerade.py From 156ca005ded0bc0abc00e9da8ef6e0eca703333c Mon Sep 17 00:00:00 2001 From: SolitudePy <47316655+SolitudePy@users.noreply.github.com> Date: Wed, 11 Jun 2025 21:56:19 +0300 Subject: [PATCH 05/14] changed staticmethods to classmethods --- .../framework/plugins/windows/malware/pebmasquerade.py | 10 ++++++---- 1 file changed, 6 insertions(+), 4 deletions(-) diff --git a/volatility3/framework/plugins/windows/malware/pebmasquerade.py b/volatility3/framework/plugins/windows/malware/pebmasquerade.py index 78676bc0d..be54300ff 100644 --- a/volatility3/framework/plugins/windows/malware/pebmasquerade.py +++ b/volatility3/framework/plugins/windows/malware/pebmasquerade.py @@ -38,8 +38,8 @@ class PebMasquerade(interfaces.plugins.PluginInterface): ), ] - @staticmethod - def _get_cmdline_image(cmdline: str) -> Union[str, PureWindowsPath]: + @classmethod + def _get_cmdline_image(cls, cmdline: str) -> Union[str, PureWindowsPath]: """Extract the executable path from a command line string. Args: @@ -71,8 +71,10 @@ class PebMasquerade(interfaces.plugins.PluginInterface): return "" - @staticmethod - def _are_paths_equal(device_path: str, drive_path: str) -> Tuple[bool, str, str]: + @classmethod + def _are_paths_equal( + cls, device_path: str, drive_path: str + ) -> Tuple[bool, str, str]: """Compare two paths to see if they are equal, ignoring drive/device root and case. Args: From 36ce047ec0e1e41a2c2340381a0d62871027e4ed Mon Sep 17 00:00:00 2001 From: SolitudePy <47316655+SolitudePy@users.noreply.github.com> Date: Wed, 11 Jun 2025 22:01:49 +0300 Subject: [PATCH 06/14] PEBMASQ: return None instead of empty string --- volatility3/framework/plugins/windows/malware/pebmasquerade.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/volatility3/framework/plugins/windows/malware/pebmasquerade.py b/volatility3/framework/plugins/windows/malware/pebmasquerade.py index be54300ff..72b138bae 100644 --- a/volatility3/framework/plugins/windows/malware/pebmasquerade.py +++ b/volatility3/framework/plugins/windows/malware/pebmasquerade.py @@ -49,7 +49,7 @@ class PebMasquerade(interfaces.plugins.PluginInterface): Union[str, PureWindowsPath]: The executable path as a string or PureWindowsPath. """ if not cmdline: - return "" + return None # Regex to extract first .exe ending string (handles quotes, paths, no quotes) match = re.search(r'(?i)(["\']?)([^"\']*?\.exe)\1(?=\s|$)', cmdline) From f79a6d6643da54777dce147a8c4af4b1d3094c7b Mon Sep 17 00:00:00 2001 From: SolitudePy <47316655+SolitudePy@users.noreply.github.com> Date: Wed, 11 Jun 2025 22:10:43 +0300 Subject: [PATCH 07/14] PEBMASQ: parameterize _generator --- .../framework/plugins/windows/malware/pebmasquerade.py | 7 ++++--- 1 file changed, 4 insertions(+), 3 deletions(-) diff --git a/volatility3/framework/plugins/windows/malware/pebmasquerade.py b/volatility3/framework/plugins/windows/malware/pebmasquerade.py index 72b138bae..2e1e47b39 100644 --- a/volatility3/framework/plugins/windows/malware/pebmasquerade.py +++ b/volatility3/framework/plugins/windows/malware/pebmasquerade.py @@ -209,8 +209,8 @@ class PebMasquerade(interfaces.plugins.PluginInterface): peb_cmdline, ) - def _generator(self): - pid_filter = pslist.PsList.create_pid_filter(self.config.get("pid", None)) + def _generator(self, pids): + pid_filter = pslist.PsList.create_pid_filter(pids) for proc in pslist.PsList.list_processes( context=self.context, @@ -359,6 +359,7 @@ class PebMasquerade(interfaces.plugins.PluginInterface): ) def run(self): + pids = self.config.get("pid", None) return renderers.TreeGrid( [ ("PID", int), @@ -369,5 +370,5 @@ class PebMasquerade(interfaces.plugins.PluginInterface): ("PEB_CommandLine_Path", str), ("Notes", str), ], - self._generator(), + self._generator(pids), ) From bcc6ce68c51a4ed6fd287813ddd84b5e800152f9 Mon Sep 17 00:00:00 2001 From: SolitudePy <47316655+SolitudePy@users.noreply.github.com> Date: Wed, 11 Jun 2025 22:13:52 +0300 Subject: [PATCH 08/14] PEBMASQ: parameterize _generator v2 --- .../framework/plugins/windows/malware/pebmasquerade.py | 10 ++++++---- 1 file changed, 6 insertions(+), 4 deletions(-) diff --git a/volatility3/framework/plugins/windows/malware/pebmasquerade.py b/volatility3/framework/plugins/windows/malware/pebmasquerade.py index 2e1e47b39..e4bd219e2 100644 --- a/volatility3/framework/plugins/windows/malware/pebmasquerade.py +++ b/volatility3/framework/plugins/windows/malware/pebmasquerade.py @@ -209,12 +209,12 @@ class PebMasquerade(interfaces.plugins.PluginInterface): peb_cmdline, ) - def _generator(self, pids): + def _generator(self, pids, context, kernel_module_name): pid_filter = pslist.PsList.create_pid_filter(pids) for proc in pslist.PsList.list_processes( - context=self.context, - kernel_module_name=self.config["kernel"], + context=context, + kernel_module_name=kernel_module_name, filter_func=pid_filter, ): proc_id = proc.UniqueProcessId @@ -360,6 +360,8 @@ class PebMasquerade(interfaces.plugins.PluginInterface): def run(self): pids = self.config.get("pid", None) + context = self.context + kernel_module_name = self.config["kernel"] return renderers.TreeGrid( [ ("PID", int), @@ -370,5 +372,5 @@ class PebMasquerade(interfaces.plugins.PluginInterface): ("PEB_CommandLine_Path", str), ("Notes", str), ], - self._generator(pids), + self._generator(pids, context, kernel_module_name), ) From 4adc3305a3883f33d3751ed4f6c21cf5fbcf5366 Mon Sep 17 00:00:00 2001 From: SolitudePy <47316655+SolitudePy@users.noreply.github.com> Date: Thu, 12 Jun 2025 20:04:49 +0300 Subject: [PATCH 09/14] remove error truncate --- .../plugins/windows/malware/pebmasquerade.py | 12 ++++++------ 1 file changed, 6 insertions(+), 6 deletions(-) diff --git a/volatility3/framework/plugins/windows/malware/pebmasquerade.py b/volatility3/framework/plugins/windows/malware/pebmasquerade.py index e4bd219e2..8c6c0b4b0 100644 --- a/volatility3/framework/plugins/windows/malware/pebmasquerade.py +++ b/volatility3/framework/plugins/windows/malware/pebmasquerade.py @@ -134,7 +134,7 @@ class PebMasquerade(interfaces.plugins.PluginInterface): vollog.warning( "Error reading EPROCESS.ImageFileName for PID %d: %s", proc.UniqueProcessId, - str(e)[:50], + str(e), ) try: @@ -156,7 +156,7 @@ class PebMasquerade(interfaces.plugins.PluginInterface): vollog.warning( "Error reading SeAuditProcessCreationInfo for PID %d: %s", proc.UniqueProcessId, - str(e)[:50], + str(e), ) try: @@ -176,7 +176,7 @@ class PebMasquerade(interfaces.plugins.PluginInterface): vollog.warning( "Error reading PEB.ImagePathName for PID %d: %s", proc.UniqueProcessId, - str(e)[:50], + str(e), ) try: @@ -192,14 +192,14 @@ class PebMasquerade(interfaces.plugins.PluginInterface): vollog.warning( "Error reading PEB.ProcessParameters.CommandLine for PID %d: %s", proc.UniqueProcessId, - str(e)[:50], + str(e), ) except (AttributeError, exceptions.InvalidAddressException): # Important for cases where PEB does not exist or is inaccessible (e.g SYSTEM process) vollog.debug("Unable to access PEB for PID %d", proc.UniqueProcessId) except Exception as e: vollog.warning( - "Error accessing PEB for PID %d: %s", proc.UniqueProcessId, str(e)[:50] + "Error accessing PEB for PID %d: %s", proc.UniqueProcessId, str(e) ) return ( @@ -245,7 +245,7 @@ class PebMasquerade(interfaces.plugins.PluginInterface): vollog.debug( "Error extracting command line path for PID %d: %s", proc_id, - str(e)[:50], + str(e), ) # Populate notes for enrichment From 32def71eb5a42d44709a2a1477194c8f2f61215a Mon Sep 17 00:00:00 2001 From: SolitudePy <47316655+SolitudePy@users.noreply.github.com> Date: Tue, 22 Jul 2025 21:34:55 +0300 Subject: [PATCH 10/14] remove notes --- .../plugins/windows/malware/pebmasquerade.py | 86 ++++--------------- 1 file changed, 16 insertions(+), 70 deletions(-) diff --git a/volatility3/framework/plugins/windows/malware/pebmasquerade.py b/volatility3/framework/plugins/windows/malware/pebmasquerade.py index 8c6c0b4b0..cc3ee73bc 100644 --- a/volatility3/framework/plugins/windows/malware/pebmasquerade.py +++ b/volatility3/framework/plugins/windows/malware/pebmasquerade.py @@ -225,14 +225,14 @@ class PebMasquerade(interfaces.plugins.PluginInterface): "Unable to access PEB for PID %d, skipping process", proc_id ) notes = [] - + peb_imagefilepath_length_check = False + peb_cmdline_length_check = False ( eprocess_imagefilename, eprocess_seaudit_imagefilename, peb_imagefilepath, peb_cmdline, ) = self.get_process_names(proc) - proc_name_for_row = eprocess_imagefilename # Extract command line executable path for rendering peb_cmdline_path_render = renderers.NotAvailableValue() @@ -248,63 +248,6 @@ class PebMasquerade(interfaces.plugins.PluginInterface): str(e), ) - # Populate notes for enrichment - if isinstance(eprocess_imagefilename, str) and isinstance( - peb_imagefilepath, str - ): - try: - peb_imagefilepath_basename = PureWindowsPath(peb_imagefilepath).name - peb_imagefilepath_truncated = peb_imagefilepath_basename[:14] - - # Compare EPROCESS.ImageFileName with PEB.ImageFilePath truncated to 15 characters - if ( - eprocess_imagefilename.lower() - != peb_imagefilepath_truncated.lower() - ): - notes.append( - f"'Potential PEB.ImageFilePath Spoofing: EPROCESS={eprocess_imagefilename};PEB={peb_imagefilepath_truncated}'" - ) - except Exception as e: - notes.append(f"ImageFilePath Comparison error: {str(e)[:30]}") - - if isinstance(eprocess_imagefilename, str) and isinstance(peb_cmdline, str): - try: - # Compare EPROCESS.ImageFileName with PEB.CommandLine executable path truncated to 15 characters - peb_cmdline_path = PebMasquerade._get_cmdline_image(peb_cmdline) - if isinstance(peb_cmdline_path, PureWindowsPath): - peb_cmdline_path = peb_cmdline_path.name - peb_cmdline_basename_truncated = peb_cmdline_path[:14] - if ( - eprocess_imagefilename.lower() - != peb_cmdline_basename_truncated.lower() - ): - notes.append( - f"'Potential PEB.CommandLine Spoofing: EPROCESS={eprocess_imagefilename};PEB={peb_cmdline_basename_truncated}'" - ) - except Exception as e: - notes.append(f"CommandLine comparison error: {str(e)}") - - if isinstance(eprocess_seaudit_imagefilename, str) and isinstance( - peb_imagefilepath, str - ): - try: - ( - are_equal, - eprocess_seaudit_normalized, - peb_imagefilepath_normalized, - ) = PebMasquerade._are_paths_equal( - device_path=eprocess_seaudit_imagefilename, - drive_path=peb_imagefilepath, - ) - if not are_equal: - notes.append( - f"'Potential PEB.ImageFilePath Spoofing (via _EPROCESS.SeAuditProcessCreationInfo): EPROCESS={eprocess_seaudit_normalized};PEB={peb_imagefilepath_normalized}'" - ) - except Exception as e: - notes.append( - f"SeAuditProcessCreationInfo comparison error: {str(e)[:30]}" - ) - if isinstance(peb_imagefilepath, str) and peb: try: @@ -319,12 +262,12 @@ class PebMasquerade(interfaces.plugins.PluginInterface): if (peb_imagefilepath_length != len(peb_imagefilepath)) or ( peb_imagefilepath_maxlength != len(peb_imagefilepath) ): - notes.append( - f"'PEB.ImageFilePath Length Mismatch: Length={peb_imagefilepath_length}, MaximumLength={peb_imagefilepath_maxlength}, Actual={len(peb_imagefilepath)}'" - ) + peb_imagefilepath_length_check = True except Exception as e: - notes.append( - f"PEB.ImageFilePath Length comparison error: {str(e)[:30]}" + vollog.warning( + "PEB.ImagePathName Length comparison error for PID %d: %s", + proc_id, + str(e), ) if isinstance(peb_cmdline, str) and peb: @@ -338,23 +281,26 @@ class PebMasquerade(interfaces.plugins.PluginInterface): if (peb_cmdline_length != len(peb_cmdline)) or ( peb_cmdline_maxlength != len(peb_cmdline) ): + peb_cmdline_length_check = True notes.append( f"'PEB.CommandLine Length Mismatch: Commandline={peb_cmdline}, Length={peb_cmdline_length}, MaximumLength={peb_cmdline_maxlength}, Actual={len(peb_cmdline)}'" ) except Exception as e: - notes.append( - f"PEB.CommandLine Length comparison error: {str(e)[:30]}" + vollog.warning( + "PEB.CommandLine Length comparison error for PID %d: %s", + proc_id, + str(e), ) yield ( 0, ( proc_id, - proc_name_for_row, eprocess_imagefilename, eprocess_seaudit_imagefilename, peb_imagefilepath, peb_cmdline_path_render, - "[" + ", ".join(notes) + "]" if notes else "OK", + peb_cmdline_length_check, + peb_imagefilepath_length_check, ), ) @@ -365,12 +311,12 @@ class PebMasquerade(interfaces.plugins.PluginInterface): return renderers.TreeGrid( [ ("PID", int), - ("ProcessName", str), ("EPROCESS_ImageFileName", str), ("EPROCESS_SeAudit_ImageFileName", str), ("PEB_ImageFilePath", str), ("PEB_CommandLine_Path", str), - ("Notes", str), + ("PEB_ImageFilePath_Spoofed", bool), + ("PEB_CommandLine_Spoofed", bool), ], self._generator(pids, context, kernel_module_name), ) From 035b60863308afd31d0d6ee543d296e856c54a13 Mon Sep 17 00:00:00 2001 From: SolitudePy <47316655+SolitudePy@users.noreply.github.com> Date: Tue, 22 Jul 2025 21:49:16 +0300 Subject: [PATCH 11/14] Plugins: pebmasquerade remove unused code --- .../plugins/windows/malware/pebmasquerade.py | 94 +------------------ 1 file changed, 3 insertions(+), 91 deletions(-) diff --git a/volatility3/framework/plugins/windows/malware/pebmasquerade.py b/volatility3/framework/plugins/windows/malware/pebmasquerade.py index cc3ee73bc..070716503 100644 --- a/volatility3/framework/plugins/windows/malware/pebmasquerade.py +++ b/volatility3/framework/plugins/windows/malware/pebmasquerade.py @@ -1,6 +1,4 @@ import logging -import re -from pathlib import PureWindowsPath from typing import List, Union, Tuple from volatility3.framework import interfaces, renderers, exceptions @@ -38,74 +36,8 @@ class PebMasquerade(interfaces.plugins.PluginInterface): ), ] - @classmethod - def _get_cmdline_image(cls, cmdline: str) -> Union[str, PureWindowsPath]: - """Extract the executable path from a command line string. - - Args: - cmdline (str): The command line string to parse. - - Returns: - Union[str, PureWindowsPath]: The executable path as a string or PureWindowsPath. - """ - if not cmdline: - return None - - # Regex to extract first .exe ending string (handles quotes, paths, no quotes) - match = re.search(r'(?i)(["\']?)([^"\']*?\.exe)\1(?=\s|$)', cmdline) - if match: - exe_path = match.group(2) - return PureWindowsPath(exe_path) - - # If no .exe found, extract the first token (handles quotes) - # Matches either "quoted string" or unquoted word - first_token_match = re.match(r'\s*(?:"([^"]+)"|\'([^\']+)\'|(\S+))', cmdline) - if first_token_match: - # Extract whichever group matched - executable = ( - first_token_match.group(1) - or first_token_match.group(2) - or first_token_match.group(3) - ) - return PureWindowsPath(executable).name + ".exe" - - return "" - - @classmethod - def _are_paths_equal( - cls, device_path: str, drive_path: str - ) -> Tuple[bool, str, str]: - """Compare two paths to see if they are equal, ignoring drive/device root and case. - - Args: - device_path (str): The device path (e.g. "\\Device\\HarddiskVolume1\\path") - drive_path (str): The drive path (e.g. "C:\\path") - - Returns: - tuple: (are_equal, device_path_without_drive, drive_path_without_drive) - - are_equal (bool): True if paths are equal, False otherwise - - device_path_without_drive (str): Device path without drive letter - - drive_path_without_drive (str): Drive path without drive letter - """ - pure_device_path = PureWindowsPath(device_path) - pure_drive_path = PureWindowsPath(drive_path) - device_parts = list(pure_device_path.parts) - drive_parts = list(pure_drive_path.parts) - - if pure_drive_path.is_absolute(): - new_drive_path = "/".join(drive_parts[1:]).lower() - new_device_path = "/".join(device_parts[3:]).lower() - else: - new_drive_path = "/".join(drive_parts[2:]).lower() - new_device_path = "/".join(device_parts[4:]).lower() - - return ( - new_drive_path == new_device_path, - new_device_path, - new_drive_path, - ) - - def get_process_names(self, proc: interfaces.objects.ObjectInterface) -> Tuple[ + @staticmethod + def get_process_names(proc: interfaces.objects.ObjectInterface) -> Tuple[ Union[str, renderers.NotAvailableValue], Union[str, renderers.NotAvailableValue], Union[str, renderers.NotAvailableValue], @@ -224,7 +156,6 @@ class PebMasquerade(interfaces.plugins.PluginInterface): vollog.debug( "Unable to access PEB for PID %d, skipping process", proc_id ) - notes = [] peb_imagefilepath_length_check = False peb_cmdline_length_check = False ( @@ -232,21 +163,7 @@ class PebMasquerade(interfaces.plugins.PluginInterface): eprocess_seaudit_imagefilename, peb_imagefilepath, peb_cmdline, - ) = self.get_process_names(proc) - - # Extract command line executable path for rendering - peb_cmdline_path_render = renderers.NotAvailableValue() - if isinstance(peb_cmdline, str): - try: - peb_cmdline_path_render = str( - PebMasquerade._get_cmdline_image(peb_cmdline) - ) - except Exception as e: - vollog.debug( - "Error extracting command line path for PID %d: %s", - proc_id, - str(e), - ) + ) = PebMasquerade.get_process_names(proc) if isinstance(peb_imagefilepath, str) and peb: try: @@ -282,9 +199,6 @@ class PebMasquerade(interfaces.plugins.PluginInterface): peb_cmdline_maxlength != len(peb_cmdline) ): peb_cmdline_length_check = True - notes.append( - f"'PEB.CommandLine Length Mismatch: Commandline={peb_cmdline}, Length={peb_cmdline_length}, MaximumLength={peb_cmdline_maxlength}, Actual={len(peb_cmdline)}'" - ) except Exception as e: vollog.warning( "PEB.CommandLine Length comparison error for PID %d: %s", @@ -298,7 +212,6 @@ class PebMasquerade(interfaces.plugins.PluginInterface): eprocess_imagefilename, eprocess_seaudit_imagefilename, peb_imagefilepath, - peb_cmdline_path_render, peb_cmdline_length_check, peb_imagefilepath_length_check, ), @@ -314,7 +227,6 @@ class PebMasquerade(interfaces.plugins.PluginInterface): ("EPROCESS_ImageFileName", str), ("EPROCESS_SeAudit_ImageFileName", str), ("PEB_ImageFilePath", str), - ("PEB_CommandLine_Path", str), ("PEB_ImageFilePath_Spoofed", bool), ("PEB_CommandLine_Spoofed", bool), ], From 2dd0bec92744d376fb649d7b1d759c5282db2cda Mon Sep 17 00:00:00 2001 From: SolitudePy <47316655+SolitudePy@users.noreply.github.com> Date: Wed, 17 Sep 2025 02:03:48 +0300 Subject: [PATCH 12/14] Plugins: pebmasq - change staticmethod to classmethod --- volatility3/framework/plugins/windows/malware/pebmasquerade.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/volatility3/framework/plugins/windows/malware/pebmasquerade.py b/volatility3/framework/plugins/windows/malware/pebmasquerade.py index 070716503..c2636820b 100644 --- a/volatility3/framework/plugins/windows/malware/pebmasquerade.py +++ b/volatility3/framework/plugins/windows/malware/pebmasquerade.py @@ -36,7 +36,7 @@ class PebMasquerade(interfaces.plugins.PluginInterface): ), ] - @staticmethod + @classmethod def get_process_names(proc: interfaces.objects.ObjectInterface) -> Tuple[ Union[str, renderers.NotAvailableValue], Union[str, renderers.NotAvailableValue], From a6d1b34d36ab8a3d63dc3d1f1f07bf965e9712dc Mon Sep 17 00:00:00 2001 From: SolitudePy <47316655+SolitudePy@users.noreply.github.com> Date: Wed, 17 Sep 2025 13:46:34 +0300 Subject: [PATCH 13/14] Plugins: pebmasq fix classmethod --- volatility3/framework/plugins/windows/malware/pebmasquerade.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/volatility3/framework/plugins/windows/malware/pebmasquerade.py b/volatility3/framework/plugins/windows/malware/pebmasquerade.py index c2636820b..3d662789a 100644 --- a/volatility3/framework/plugins/windows/malware/pebmasquerade.py +++ b/volatility3/framework/plugins/windows/malware/pebmasquerade.py @@ -37,7 +37,7 @@ class PebMasquerade(interfaces.plugins.PluginInterface): ] @classmethod - def get_process_names(proc: interfaces.objects.ObjectInterface) -> Tuple[ + def get_process_names(cls, proc: interfaces.objects.ObjectInterface) -> Tuple[ Union[str, renderers.NotAvailableValue], Union[str, renderers.NotAvailableValue], Union[str, renderers.NotAvailableValue], From 0e17057b0b6a01425cf3f4c66c42233e7ccb7f83 Mon Sep 17 00:00:00 2001 From: SolitudePy <47316655+SolitudePy@users.noreply.github.com> Date: Tue, 23 Sep 2025 21:16:47 +0300 Subject: [PATCH 14/14] Plugins - pebmasq required framework version --- volatility3/framework/plugins/windows/malware/pebmasquerade.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/volatility3/framework/plugins/windows/malware/pebmasquerade.py b/volatility3/framework/plugins/windows/malware/pebmasquerade.py index 3d662789a..dd85fb570 100644 --- a/volatility3/framework/plugins/windows/malware/pebmasquerade.py +++ b/volatility3/framework/plugins/windows/malware/pebmasquerade.py @@ -15,7 +15,7 @@ class PebMasquerade(interfaces.plugins.PluginInterface): """Detects potential process name spoofing by comparing EPROCESS and PEB data.""" _version = (1, 0, 0) - _required_framework_version = (2, 0, 0) + _required_framework_version = (2, 27, 0) @classmethod def get_requirements(cls) -> List[interfaces.configuration.RequirementInterface]: