Merge pull request #1475 from Pennyw0rth/neff-improve-loadtime

nxc/connection.py

@@ -19,11 +19,6 @@ from nxc.loaders.moduleloader import ModuleLoader, ModuleOptionsError
 from nxc.logger import nxc_logger, NXCAdapter
 from nxc.context import Context
 from nxc.paths import NXC_PATH
-from nxc.protocols.ldap.laps import laps_search
-from nxc.helpers.pfx import pfx_auth
-
-from impacket.dcerpc.v5 import transport
-from impacket.krb5.ccache import CCache
 
 sem = BoundedSemaphore(1)
 fail_lock = Lock()
@@ -101,37 +96,6 @@ def requires_admin(func):
     return wraps(func)(_decorator)
 
 
-def dcom_FirewallChecker(iInterface, remoteHost, timeout):
-    stringBindings = iInterface.get_cinstance().get_string_bindings()
-    for strBinding in stringBindings:
-        if strBinding["wTowerId"] == 7:
-            if strBinding["aNetworkAddr"].find("[") >= 0:
-                binding, _, bindingPort = strBinding["aNetworkAddr"].partition("[")
-                bindingPort = "[" + bindingPort
-            else:
-                binding = strBinding["aNetworkAddr"]
-                bindingPort = ""
-
-            if binding.upper().find(iInterface.get_target().upper()) >= 0:
-                stringBinding = "ncacn_ip_tcp:" + strBinding["aNetworkAddr"][:-1]
-                break
-            elif iInterface.is_fqdn() and binding.upper().find(iInterface.get_target().upper().partition(".")[0]) >= 0:
-                stringBinding = f"ncacn_ip_tcp:{iInterface.get_target()}{bindingPort}"
-    if "stringBinding" not in locals():
-        return True, None
-    try:
-        rpctransport = transport.DCERPCTransportFactory(stringBinding)
-        rpctransport.setRemoteHost(remoteHost)
-        rpctransport.set_connect_timeout(timeout)
-        rpctransport.connect()
-        rpctransport.disconnect()
-    except Exception as e:
-        nxc_logger.debug(f"Exception while connecting to {stringBinding}: {e}")
-        return False, stringBinding
-    else:
-        return True, stringBinding
-
-
 class connection:
     def __init__(self, args, db, target):
         self.args = args
@@ -560,6 +524,7 @@ class connection:
         if self.args.use_kcache:
             self.logger.debug("Trying to authenticate using Kerberos cache")
             with sem:
+                from impacket.krb5.ccache import CCache
                 username = self.args.username[0] if len(self.args.username) else CCache.parseFile()[1]
                 password = self.args.password[0] if len(self.args.password) else ""
                 self.kerberos_login(self.domain, username, password, "", "", self.kdcHost, True)
@@ -576,10 +541,12 @@ class connection:
                 self.logger.fail("You must specify a username when using certificate authentication")
                 return False
             with sem:
+                from nxc.helpers.pfx import pfx_auth
                 return pfx_auth(self)
 
         if hasattr(self.args, "laps") and self.args.laps:
             self.logger.debug("Trying to authenticate using LAPS")
+            from nxc.protocols.ldap.laps import laps_search
             username[0], secret[0], domain[0] = laps_search(self, username, secret, cred_type, domain, self.dns_server)
             cred_type = ["plaintext"]
             if not (username[0] or secret[0] or domain[0]):

nxc/helpers/rpc.py

@@ -99,3 +99,34 @@ class NXCRPCConnection:
 
         self.setup_credentials(rpc_transport, anonymous_rpc)
         return rpc_transport
+
+
+def dcom_FirewallChecker(iInterface, remoteHost, timeout, logger):
+    stringBindings = iInterface.get_cinstance().get_string_bindings()
+    for strBinding in stringBindings:
+        if strBinding["wTowerId"] == 7:
+            if strBinding["aNetworkAddr"].find("[") >= 0:
+                binding, _, bindingPort = strBinding["aNetworkAddr"].partition("[")
+                bindingPort = "[" + bindingPort
+            else:
+                binding = strBinding["aNetworkAddr"]
+                bindingPort = ""
+
+            if binding.upper().find(iInterface.get_target().upper()) >= 0:
+                stringBinding = "ncacn_ip_tcp:" + strBinding["aNetworkAddr"][:-1]
+                break
+            elif iInterface.is_fqdn() and binding.upper().find(iInterface.get_target().upper().partition(".")[0]) >= 0:
+                stringBinding = f"ncacn_ip_tcp:{iInterface.get_target()}{bindingPort}"
+    if "stringBinding" not in locals():
+        return True, None
+    try:
+        rpctransport = transport.DCERPCTransportFactory(stringBinding)
+        rpctransport.setRemoteHost(remoteHost)
+        rpctransport.set_connect_timeout(timeout)
+        rpctransport.connect()
+        rpctransport.disconnect()
+    except Exception as e:
+        logger.debug(f"Exception while connecting to {stringBinding}: {e}")
+        return False, stringBinding
+    else:
+        return True, stringBinding

nxc/modules/rdp.py

@@ -1,6 +1,6 @@
 from sys import exit
 
-from nxc.connection import dcom_FirewallChecker
+from nxc.helpers.rpc import dcom_FirewallChecker
 from nxc.helpers.misc import CATEGORY
 
 from impacket.dcerpc.v5 import rrp
@@ -248,7 +248,7 @@ class RdpWmi:
 
             i_interface = self.__dcom.CoCreateInstanceEx(wmi.CLSID_WbemLevel1Login, wmi.IID_IWbemLevel1Login)
             if self.__currentprotocol == "smb":
-                flag, self.__stringBinding = dcom_FirewallChecker(i_interface, self.__remoteHost, self.__timeout)
+                flag, self.__stringBinding = dcom_FirewallChecker(i_interface, self.__remoteHost, self.__timeout, self.logger)
                 if not flag or not self.__stringBinding:
                     error_msg = f'RDP-WMI: Dcom initialization failed on connection with stringbinding: "{self.__stringBinding}", please increase the timeout with the module option "DCOM-TIMEOUT=10". If it\'s still failing maybe something is blocking the RPC connection, please try to use "-o" with "METHOD=smb"'
 

nxc/protocols/smb.py

@@ -61,7 +61,8 @@ from impacket.smb3structs import (
 from impacket.dcerpc.v5 import tsts as TSTS
 
 from nxc.config import process_secret, host_info_colors, check_guest_account, display_dc
-from nxc.connection import connection, sem, requires_admin, dcom_FirewallChecker
+from nxc.connection import connection, sem, requires_admin
+from nxc.helpers.rpc import dcom_FirewallChecker
 from nxc.helpers.misc import gen_random_string, validate_ntlm
 from nxc.logger import NXCAdapter
 from nxc.paths import NXC_PATH
@@ -1991,7 +1992,7 @@ class smb(connection):
         try:
             dcom = DCOMConnection(self.remoteName, self.username, self.password, self.domain, self.lmhash, self.nthash, oxidResolver=True, doKerberos=self.kerberos, kdcHost=self.kdcHost, aesKey=self.aesKey, remoteHost=self.host)
             iInterface = dcom.CoCreateInstanceEx(CLSID_WbemLevel1Login, IID_IWbemLevel1Login)
-            flag, stringBinding = dcom_FirewallChecker(iInterface, self.host, self.args.dcom_timeout)
+            flag, stringBinding = dcom_FirewallChecker(iInterface, self.host, self.args.dcom_timeout, self.logger)
             if not flag or not stringBinding:
                 error_msg = f"WMI Query: Dcom initialization failed on connection with stringbinding: '{stringBinding}', please increase the timeout with the option '--dcom-timeout'. If it's still failing maybe something is blocking the RPC connection, try another exec method"
 

nxc/protocols/smb/mmcexec.py

@@ -27,7 +27,7 @@
 
 from os.path import join as path_join
 from time import sleep
-from nxc.connection import dcom_FirewallChecker
+from nxc.helpers.rpc import dcom_FirewallChecker
 from nxc.helpers.misc import gen_random_string
 from nxc.paths import TMP_PATH
 
@@ -108,7 +108,7 @@ class MMCEXEC:
             self.logger.info(f"Got Exception while connecting with DCOM: {e}")
             # Make it force break function
             self.__dcom.disconnect()
-        flag, self.__stringBinding = dcom_FirewallChecker(iInterface, self.__remoteHost, self.__timeout)
+        flag, self.__stringBinding = dcom_FirewallChecker(iInterface, self.__remoteHost, self.__timeout, self.logger)
         if not flag or not self.__stringBinding:
             error_msg = f'MMCEXEC: Dcom initialization failed on connection with stringbinding: "{self.__stringBinding}", please increase the timeout with the option "--dcom-timeout". If it\'s still failing maybe something is blocking the RPC connection, try another exec method'
 

nxc/protocols/smb/wmiexec.py

@@ -1,7 +1,7 @@
 import ntpath
 import os
 from time import sleep
-from nxc.connection import dcom_FirewallChecker
+from nxc.helpers.rpc import dcom_FirewallChecker
 from nxc.helpers.misc import gen_random_string
 from nxc.paths import TMP_PATH
 from impacket.dcerpc.v5.dcomrt import DCOMConnection
@@ -57,7 +57,7 @@ class WMIEXEC:
             remoteHost=self.__remoteHost,
         )
         iInterface = self.__dcom.CoCreateInstanceEx(wmi.CLSID_WbemLevel1Login, wmi.IID_IWbemLevel1Login)
-        flag, self.__stringBinding = dcom_FirewallChecker(iInterface, self.__remoteHost, self.__timeout)
+        flag, self.__stringBinding = dcom_FirewallChecker(iInterface, self.__remoteHost, self.__timeout, self.logger)
         if not flag or not self.__stringBinding:
             error_msg = f'WMIEXEC: Dcom initialization failed on connection with stringbinding: "{self.__stringBinding}", please increase the timeout with the option "--dcom-timeout". If it\'s still failing maybe something is blocking the RPC connection, try another exec method'
 

nxc/protocols/wmi.py

@@ -9,7 +9,8 @@ import base64
 from nxc.helpers.negotiate_parser import parse_challenge
 from nxc.helpers.misc import sanitize_dns
 from nxc.config import process_secret
-from nxc.connection import connection, dcom_FirewallChecker, requires_admin
+from nxc.connection import connection, requires_admin
+from nxc.helpers.rpc import dcom_FirewallChecker
 from nxc.helpers.misc import validate_ntlm
 from nxc.logger import NXCAdapter
 from nxc.helpers.logger import highlight
@@ -180,7 +181,7 @@ class wmi(connection):
         try:
             self.dcom_conn = DCOMConnection(self.remoteName, self.username, self.password, self.domain, self.lmhash, self.nthash, oxidResolver=True, doKerberos=self.doKerberos, kdcHost=self.kdcHost, aesKey=self.aesKey, remoteHost=self.host)
             iInterface = self.dcom_conn.CoCreateInstanceEx(CLSID_WbemLevel1Login, IID_IWbemLevel1Login)
-            flag, self.stringBinding = dcom_FirewallChecker(iInterface, self.host, self.args.rpc_timeout)
+            flag, self.stringBinding = dcom_FirewallChecker(iInterface, self.host, self.args.rpc_timeout, self.logger)
         except Exception as e:
             self.logger.debug(f"Received error while checking admin: {e}")
             if "access_denied" not in str(e).lower():