Add RPC helper class and migrate SMB/module callers

nxc/helpers/rpc.py

@@ -0,0 +1,97 @@
+# DCE/RPC over SMB/TCP using the NXC connection's creds; reuse SMB when already
+# logged in so Kerberos delegation (e.g. S4U2Proxy ST) applies to pipe RPC too.
+import contextlib
+
+from impacket.dcerpc.v5 import transport
+from impacket.dcerpc.v5.rpcrt import RPC_C_AUTHN_GSS_NEGOTIATE
+
+
+class NXCRPCConnection:
+    def __init__(self, connection, force_tcp=False):
+        self.connection = connection
+        self.force_tcp = force_tcp
+        self.dce = None
+        self.rpc_transport = None
+
+    def connect(self, named_pipe, interface_uuid, target_ip=None, auth_level=None, string_binding=None, set_remote_host=None, anonymous_rpc=False):
+        if string_binding is not None:
+            self.rpc_transport = self.create_from_string_binding(string_binding, target_ip, set_remote_host, anonymous_rpc)
+        elif self.force_tcp:
+            self.rpc_transport = self.create_tcp_transport(target_ip, anonymous_rpc)
+        else:
+            self.rpc_transport = self.create_smb_transport(named_pipe, target_ip)
+
+        self.dce = self.rpc_transport.get_dce_rpc()
+
+        if self.connection.kerberos and not anonymous_rpc:
+            self.dce.set_auth_type(RPC_C_AUTHN_GSS_NEGOTIATE)
+
+        self.dce.connect()
+
+        if auth_level is not None:
+            self.dce.set_auth_level(auth_level)
+
+        self.dce.bind(interface_uuid)
+        return self.dce
+
+    def disconnect(self):
+        if self.dce:
+            with contextlib.suppress(Exception):
+                self.dce.disconnect()
+            self.dce = None
+
+    @property
+    def transport(self):
+        return self.rpc_transport
+
+    def get_smb_connection(self):
+        if self.rpc_transport and hasattr(self.rpc_transport, "get_smb_connection"):
+            return self.rpc_transport.get_smb_connection()
+        return None
+
+    def __enter__(self):
+        return self
+
+    def __exit__(self, exc_type, exc_val, exc_tb):
+        self.disconnect()
+        return False
+
+    def setup_credentials(self, rpc_transport, anonymous_rpc=False):
+        conn = self.connection
+        if anonymous_rpc:
+            rpc_transport.set_credentials("", "", "", "", "", "")
+            rpc_transport.set_kerberos(False, None)
+        else:
+            rpc_transport.set_credentials(conn.username, conn.password if conn.password else "", conn.domain, conn.lmhash, conn.nthash, conn.aesKey)
+            if conn.kerberos:
+                rpc_transport.set_kerberos(conn.kerberos, conn.kdcHost)
+
+    def create_smb_transport(self, named_pipe, target_ip=None):
+        conn = self.connection
+
+        if conn.conn is not None:
+            return transport.SMBTransport(conn.conn.getRemoteHost(), filename=named_pipe, smb_connection=conn.conn,)
+        result = transport.SMBTransport(conn.remoteName, conn.port, named_pipe, conn.username, conn.password if conn.password else "", conn.domain, conn.lmhash, conn.nthash, conn.aesKey, doKerberos=conn.kerberos, kdcHost=conn.kdcHost)
+        if target_ip or conn.host:
+            result.setRemoteHost(target_ip or conn.host)
+        return result
+
+    def create_tcp_transport(self, target_ip=None, anonymous_rpc=False):
+        conn = self.connection
+        target = target_ip or conn.remoteName
+
+        result = transport.DCERPCTransportFactory(rf"ncacn_ip_tcp:{target}")
+        result.setRemoteHost(target)
+        self.setup_credentials(result, anonymous_rpc)
+        return result
+
+    def create_from_string_binding(self, string_binding, target_ip=None, set_remote_host=None, anonymous_rpc=False):
+        result = transport.DCERPCTransportFactory(string_binding)
+
+        if target_ip is not None:
+            result.setRemoteHost(target_ip)
+        elif set_remote_host is not None:
+            result.setRemoteHost(set_remote_host)
+
+        self.setup_credentials(result, anonymous_rpc)
+        return result

nxc/modules/add-computer.py

@@ -1,8 +1,9 @@
 import sys
 
-from impacket.dcerpc.v5 import samr, transport
+from impacket.dcerpc.v5 import samr
 from impacket.ldap.ldap import LDAPSessionError, MODIFY_REPLACE
 from nxc.helpers.misc import CATEGORY
+from nxc.helpers.rpc import NXCRPCConnection
 
 
 class NXCModule:
@@ -81,12 +82,9 @@ def _db_add_credential(self):
 
     def _do_samr(self):
         conn = self.connection
-        rpc_transport = transport.SMBTransport(conn.conn.getRemoteHost(), 445, r"\samr", smb_connection=conn.conn)
 
         try:
-            dce = rpc_transport.get_dce_rpc()
-            dce.connect()
-            dce.bind(samr.MSRPC_UUID_SAMR)
+            dce = NXCRPCConnection(conn).connect(r"\samr", samr.MSRPC_UUID_SAMR)
         except Exception as e:
             self.context.log.fail(f"Failed to connect to SAMR: {e}")
             return

nxc/modules/backup_operator.py

@@ -3,9 +3,9 @@
 
 from impacket.examples.secretsdump import SAMHashes, LSASecrets, LocalOperations
 from impacket.smbconnection import SessionError
-from impacket.dcerpc.v5 import transport, rrp
-from impacket.dcerpc.v5.rpcrt import RPC_C_AUTHN_GSS_NEGOTIATE
+from impacket.dcerpc.v5 import rrp
 from nxc.helpers.misc import CATEGORY, gen_random_string
+from nxc.helpers.rpc import NXCRPCConnection
 
 
 class NXCModule:
@@ -31,17 +31,9 @@ def on_login(self, context, connection):
         # enable remote registry
         context.log.display("Triggering RemoteRegistry to start through named pipe...")
         connection.trigger_winreg()
-        rpc = transport.DCERPCTransportFactory(r"ncacn_np:445[\pipe\winreg]")
-        rpc.set_smb_connection(connection.conn)
-        if connection.kerberos:
-            rpc.set_kerberos(connection.kerberos, kdcHost=connection.kdcHost)
-        dce = rpc.get_dce_rpc()
-        if connection.kerberos:
-            dce.set_auth_type(RPC_C_AUTHN_GSS_NEGOTIATE)
+        dce = NXCRPCConnection(connection).connect(r"\winreg", rrp.MSRPC_UUID_RRP)
 
         try:
-            dce.connect()
-            dce.bind(rrp.MSRPC_UUID_RRP)
 
             for hive in ["HKLM\\SAM", "HKLM\\SYSTEM", "HKLM\\SECURITY"]:
                 hRootKey, subKey = self._strip_root_key(dce, hive)

nxc/modules/change-password.py

@@ -1,7 +1,8 @@
 import sys
-from impacket.dcerpc.v5 import samr, epm, transport
+from impacket.dcerpc.v5 import samr, epm
 from impacket.dcerpc.v5.rpcrt import DCERPCException
 from nxc.helpers.misc import CATEGORY
+from nxc.helpers.rpc import NXCRPCConnection
 
 
 class NXCModule:
@@ -45,31 +46,21 @@ def options(self, context, module_options):
     def authenticate(self, context, connection, protocol, anonymous=False):
         # Authenticate to the target using DCE/RPC with either user credentials or a null session. Establishes a connection and binds to the SAMR service.
         try:
-            # Map to the SAMR endpoint on the target
-            string_binding = epm.hept_map(connection.host, samr.MSRPC_UUID_SAMR, protocol=protocol)
-            rpctransport = transport.DCERPCTransportFactory(string_binding)
-            rpctransport.setRemoteHost(connection.host)
-
             if anonymous:
-                rpctransport.set_credentials("", "", "", "", "", "")
-                rpctransport.set_kerberos(False, None)
+                string_binding = epm.hept_map(connection.host, samr.MSRPC_UUID_SAMR, protocol=protocol)
+                dce = NXCRPCConnection(connection).connect(
+                    None,
+                    samr.MSRPC_UUID_SAMR,
+                    string_binding=string_binding,
+                    set_remote_host=connection.host,
+                    anonymous_rpc=True,
+                )
                 context.log.info("Connecting with null session credentials.")
             else:
-                rpctransport.set_credentials(
-                    connection.username,
-                    connection.password,
-                    connection.domain,
-                    connection.lmhash,
-                    connection.nthash,
-                    aesKey=connection.aesKey,
-                )
+                dce = NXCRPCConnection(connection).connect(r"\samr", samr.MSRPC_UUID_SAMR)
                 context.log.info(f"Connecting as {connection.domain}\\{connection.username}")
 
-            # Connect to the DCE/RPC endpoint and bind to the SAMR service
-            dce = rpctransport.get_dce_rpc()
-            dce.connect()
             context.log.info("[+] Successfully connected to DCE/RPC")
-            dce.bind(samr.MSRPC_UUID_SAMR)
             context.log.info("[+] Successfully bound to SAMR")
             return dce
         except DCERPCException as e:

nxc/modules/enum_av.py

@@ -3,11 +3,11 @@
 # Module by @mpgn_x64
 # https://twitter.com/mpgn_x64
 
-from impacket.dcerpc.v5 import lsat, lsad, transport
+from impacket.dcerpc.v5 import lsat, lsad
 from impacket.dcerpc.v5.dtypes import NULL, MAXIMUM_ALLOWED, RPC_UNICODE_STRING
-from impacket.dcerpc.v5.rpcrt import RPC_C_AUTHN_GSS_NEGOTIATE
 import pathlib
 from nxc.helpers.misc import CATEGORY
+from nxc.helpers.rpc import NXCRPCConnection
 
 
 class NXCModule:
@@ -45,20 +45,9 @@ def _detect_installed_services(self, context, connection, target):
         results = {}
 
         try:
-            lsa = LsaLookupNames(
-                domain=connection.domain,
-                username=connection.username,
-                password=connection.password,
-                remote_name=target,
-                do_kerberos=connection.kerberos,
-                remoteHost=connection.host,
-                kdcHost=connection.kdcHost,
-                lmhash=connection.lmhash,
-                nthash=connection.nthash,
-                aesKey=connection.aesKey
-            )
+            lsa = LsaLookupNames(connection=connection)
 
-            dce, _ = lsa.connect()
+            dce = lsa.connect()
             policyHandle = lsa.open_policy(dce)
             for product in conf["products"]:
                 for service in product["services"]:
@@ -103,82 +92,11 @@ def dump_results(self, results, context):
 
 
 class LsaLookupNames:
-    timeout = None
-    authn_level = None
-    protocol = None
-    transfer_syntax = None
-    machine_account = False
+    def __init__(self, connection=None):
+        self.connection = connection
 
-    iface_uuid = lsat.MSRPC_UUID_LSAT
-    authn = True
-
-    def __init__(
-        self,
-        domain="",
-        username="",
-        password="",
-        remote_name="",
-        do_kerberos=False,
-        remoteHost="",
-        kdcHost="",
-        lmhash="",
-        nthash="",
-        aesKey="",
-    ):
-        self.domain = domain
-        self.username = username
-        self.password = password
-        self.remoteName = remote_name
-        self.string_binding = rf"ncacn_np:{remote_name}[\PIPE\lsarpc]"
-        self.doKerberos = do_kerberos
-        self.lmhash = lmhash
-        self.nthash = nthash
-        self.aesKey = aesKey
-        self.kdcHost = kdcHost
-        self.remoteHost = remoteHost
-
-    def connect(self, string_binding=None, iface_uuid=None):
-        """Obtains a RPC Transport and a DCE interface according to the bindings and
-        transfer syntax specified.
-        :return: tuple of DCE/RPC and RPC Transport objects
-        :rtype: (DCERPC_v5, DCERPCTransport)
-        """
-        string_binding = string_binding or self.string_binding
-        if not string_binding:
-            raise NotImplementedError("String binding must be defined")
-
-        rpc_transport = transport.DCERPCTransportFactory(string_binding)
-        rpc_transport.setRemoteHost(self.remoteHost)
-
-        # Set timeout if defined
-        if self.timeout:
-            rpc_transport.set_connect_timeout(self.timeout)
-
-        # Authenticate if specified
-        if self.authn and hasattr(rpc_transport, "set_credentials"):
-            # This method exists only for selected protocol sequences.
-            rpc_transport.set_credentials(self.username, self.password, self.domain, self.lmhash, self.nthash, self.aesKey)
-
-        if self.doKerberos:
-            rpc_transport.set_kerberos(self.doKerberos, kdcHost=self.kdcHost)
-
-        # Gets the DCE RPC object
-        dce = rpc_transport.get_dce_rpc()
-
-        if self.doKerberos:
-            dce.set_auth_type(RPC_C_AUTHN_GSS_NEGOTIATE)
-
-        # Connect
-        dce.connect()
-
-        # Bind if specified
-        iface_uuid = iface_uuid or self.iface_uuid
-        if iface_uuid and self.transfer_syntax:
-            dce.bind(iface_uuid, transfer_syntax=self.transfer_syntax)
-        elif iface_uuid:
-            dce.bind(iface_uuid)
-
-        return dce, rpc_transport
+    def connect(self):
+        return NXCRPCConnection(self.connection).connect(r"\lsarpc", lsat.MSRPC_UUID_LSAT)
 
     def open_policy(self, dce):
         request = lsad.LsarOpenPolicy2()

nxc/modules/enum_ca.py

@@ -7,6 +7,8 @@
 from impacket import uuid
 import requests
 from nxc.helpers.misc import CATEGORY
+from nxc.helpers.rpc import NXCRPCConnection
+import contextlib
 
 
 class NXCModule:
@@ -68,8 +70,18 @@ def on_login(self, context, connection):
         else:
             rpctransport.setRemoteHost(connection.host)
 
+        dce = None
         try:
-            entries = self.__fetchList(connection.kerberos, rpctransport)
+            if int(self.__port) == 135:
+                rpc = NXCRPCConnection(connection, force_tcp=True)
+                rpc.rpc_transport = rpc.create_tcp_transport(target_ip=connection.host)
+                dce = rpc.rpc_transport.get_dce_rpc()
+                if connection.kerberos:
+                    dce.set_auth_type(RPC_C_AUTHN_GSS_NEGOTIATE)
+                dce.connect()
+                entries = list(epm.hept_lookup(None, dce=dce))
+            else:
+                entries = self.__fetchList(connection.kerberos, rpctransport)
         except Exception as e:
             error_text = f"Protocol failed: {e}"
             context.log.fail(error_text)
@@ -81,6 +93,10 @@ def on_login(self, context, connection):
                 context.log.fail("This usually means the target does not allow "
                                  "to connect to its epmapper using RpcProxy.")
             return
+        finally:
+            if dce is not None:
+                with contextlib.suppress(Exception):
+                    dce.disconnect()
 
         for entry in entries:
             tmpUUID = str(entry["tower"]["Floors"][0])