Remove smb from ldap proto

Author: mpgn

nxc/protocols/ldap.py

@@ -31,8 +31,8 @@
 from impacket.ldap import ldaptypes
 from impacket.ldap import ldapasn1 as ldapasn1_impacket
 from impacket.ldap.ldap import LDAPFilterSyntaxError
-from impacket.smb import SMB_DIALECT
 from impacket.smbconnection import SMBConnection, SessionError
+from impacket.ntlm import getNTLMSSPType1
 
 from nxc.config import process_secret, host_info_colors
 from nxc.connection import connection
@@ -42,6 +42,7 @@
 from nxc.protocols.ldap.gmsa import MSDS_MANAGEDPASSWORD_BLOB
 from nxc.protocols.ldap.kerberos import KerberosAttacks
 from nxc.parsers.ldap_results import parse_result_attributes
+from nxc.helpers.ntlm_parser import parse_challenge
 
 ldap_error_status = {
     "1": "STATUS_NOT_SUPPORTED",
@@ -163,25 +164,25 @@ def proto_logger(self):
             }
         )
 
-    def get_ldap_info(self, host):
+    def create_conn_obj(self):
         try:
             proto = "ldaps" if (self.args.gmsa or self.port == 636) else "ldap"
-            ldap_url = f"{proto}://{host}"
+            ldap_url = f"{proto}://{self.host}"
             self.logger.info(f"Connecting to {ldap_url} with no baseDN")
             try:
-                ldap_connection = ldap_impacket.LDAPConnection(ldap_url, dstIp=self.host)
-                if ldap_connection:
-                    self.logger.debug(f"ldap_connection: {ldap_connection}")
+                self.ldap_connection = ldap_impacket.LDAPConnection(ldap_url, dstIp=self.host)
+                if self.ldap_connection:
+                    self.logger.debug(f"ldap_connection: {self.ldap_connection}")
             except SysCallError as e:
                 if proto == "ldaps":
                     self.logger.fail(f"LDAPs connection to {ldap_url} failed - {e}")
                     # https://learn.microsoft.com/en-us/troubleshoot/windows-server/identity/enable-ldap-over-ssl-3rd-certification-authority
                     self.logger.fail("Even if the port is open, LDAPS may not be configured")
                 else:
                     self.logger.fail(f"LDAP connection to {ldap_url} failed: {e}")
-                exit(1)
+                return False
 
-            resp = ldap_connection.search(
+            resp = self.ldap_connection.search(
                 scope=ldapasn1_impacket.Scope("baseObject"),
                 attributes=["defaultNamingContext", "dnsHostName"],
                 sizeLimit=0,
@@ -208,42 +209,18 @@ def get_ldap_info(self, host):
                     self.logger.debug("Exception:", exc_info=True)
                     self.logger.info(f"Skipping item, cannot process due to error {e}")
         except OSError:
-            return [None, None, None]
+            return False
         self.logger.debug(f"Target: {target}; target_domain: {target_domain}; base_dn: {base_dn}")
-        return [target, target_domain, base_dn]
-
-    def get_os_arch(self):
-        try:
-            string_binding = rf"ncacn_ip_tcp:{self.host}[135]"
-            transport = DCERPCTransportFactory(string_binding)
-            transport.setRemoteHost(self.host)
-            transport.set_connect_timeout(5)
-            dce = transport.get_dce_rpc()
-            if self.args.kerberos:
-                dce.set_auth_type(RPC_C_AUTHN_GSS_NEGOTIATE)
-            dce.connect()
-            try:
-                dce.bind(
-                    MSRPC_UUID_PORTMAP,
-                    transfer_syntax=("71710533-BEBA-4937-8319-B5DBEF9CCC36", "1.0"),
-                )
-            except DCERPCException as e:
-                if str(e).find("syntaxes_not_supported") >= 0:
-                    dce.disconnect()
-                    return 32
-            else:
-                dce.disconnect()
-                return 64
-        except Exception as e:
-            self.logger.fail(f"Error retrieving os arch of {self.host}: {e!s}")
-
-        return 0
+        self.target = target
+        self.targetDomain = target_domain
+        self.baseDN = base_dn
+        return True
 
     def get_ldap_username(self):
         extended_request = ldapasn1_impacket.ExtendedRequest()
         extended_request["requestName"] = "1.3.6.1.4.1.4203.1.11.3"  # whoami
 
-        response = self.ldapConnection.sendReceive(extended_request)
+        response = self.ldap_connection.sendReceive(extended_request)
         for message in response:
             search_result = message["protocolOp"].getComponent()
             if search_result["resultCode"] == ldapasn1_impacket.ResultCode("success"):
@@ -254,46 +231,26 @@ def get_ldap_username(self):
         return ""
 
     def enum_host_info(self):
-        self.target, self.targetDomain, self.baseDN = self.get_ldap_info(self.host)
         self.baseDN = self.args.base_dn if self.args.base_dn else self.baseDN   # Allow overwriting baseDN from args
         self.hostname = self.target
         self.remoteName = self.target
         self.domain = self.targetDomain
-        # smb no open, specify the domain
-        if not self.args.no_smb:
-            self.local_ip = self.conn.getSMBServer().get_socket().getsockname()[0]
 
-            try:
-                self.conn.login("", "")
-            except BrokenPipeError as e:
-                self.logger.fail(f"Broken Pipe Error while attempting to login: {e}")
-            except Exception as e:
-                if "STATUS_NOT_SUPPORTED" in str(e):
-                    self.no_ntlm = True
-            if not self.no_ntlm:
-                self.hostname = self.conn.getServerName()
-                self.targetDomain = self.domain = self.conn.getServerDNSDomainName()
-            self.server_os = self.conn.getServerOS()
-            self.signing = self.conn.isSigningRequired() if self.smbv1 else self.conn._SMBConnection._Connection["RequireSigning"]
-            self.os_arch = self.get_os_arch()
-            self.logger.extra["hostname"] = self.hostname
-
-            if not self.domain:
-                self.domain = self.hostname
-            if self.args.domain:
-                self.domain = self.args.domain
-            if self.args.local_auth:
-                self.domain = self.hostname
-            self.remoteName = self.host if not self.kerberos else f"{self.hostname}.{self.domain}"
-
-            try:  # noqa: SIM105
-                # DC's seem to want us to logoff first, windows workstations sometimes reset the connection
-                self.conn.logoff()
-            except Exception:
-                pass
-
-            # Re-connect since we logged off
-            self.create_conn_obj()
+        ntlm_challenge = None
+        bindRequest = ldapasn1_impacket.BindRequest()
+        bindRequest['version'] = 3
+        bindRequest['name'] = ""
+        negotiate = getNTLMSSPType1()
+        bindRequest['authentication']['sicilyNegotiate'] = negotiate.getData()
+        try:
+            response = self.ldap_connection.sendReceive(bindRequest)[0]['protocolOp']
+            ntlm_challenge = bytes(response['bindResponse']['matchedDN'])
+        except Exception as e:
+            self.logger.debug(f"Failed to get target {self.host} ntlm challenge, error: {e!s}")
+
+        if ntlm_challenge:
+            ntlm_info = parse_challenge(ntlm_challenge)
+            self.server_os = ntlm_info["os_version"]
 
         if not self.kdcHost and self.domain:
             result = self.resolver(self.domain)
@@ -304,17 +261,10 @@ def enum_host_info(self):
 
     def print_host_info(self):
         self.logger.debug("Printing host info for LDAP")
-        if self.args.no_smb:
-            self.logger.extra["protocol"] = "LDAP" if self.port == 389 else "LDAPS"
-            self.logger.extra["port"] = self.port
-            self.logger.display(f'{self.baseDN} (Hostname: {self.hostname.split(".")[0]}) (domain: {self.domain})')
-        else:
-            self.logger.extra["protocol"] = "SMB" if not self.no_ntlm else "LDAP"
-            self.logger.extra["port"] = "445" if not self.no_ntlm else "389"
-            signing = colored(f"signing:{self.signing}", host_info_colors[0], attrs=["bold"]) if self.signing else colored(f"signing:{self.signing}", host_info_colors[1], attrs=["bold"])
-            smbv1 = colored(f"SMBv1:{self.smbv1}", host_info_colors[2], attrs=["bold"]) if self.smbv1 else colored(f"SMBv1:{self.smbv1}", host_info_colors[3], attrs=["bold"])
-            self.logger.display(f"{self.server_os}{f' x{self.os_arch}' if self.os_arch else ''} (name:{self.hostname}) (domain:{self.targetDomain}) ({signing}) ({smbv1})")
-            self.logger.extra["protocol"] = "LDAP"
+        self.logger.extra["protocol"] = "LDAP" if str(self.port) == "389" else "LDAPS"
+        self.logger.extra["port"] = self.port
+        self.logger.extra["hostname"] = self.target.split(".")[0].upper()
+        self.logger.display(f"{self.server_os} (name:{self.hostname}) (domain:{self.domain})")
 
     def kerberos_login(self, domain, username, password="", ntlm_hash="", aesKey="", kdcHost="", useCache=False):
         self.username = username
@@ -594,40 +544,6 @@ def hash_login(self, domain, username, ntlm_hash):
             self.logger.fail(f"{self.domain}\\{self.username}:{process_secret(self.password)} {'Error connecting to the domain, are you sure LDAP service is running on the target?'} \nError: {e}")
             return False
 
-    def create_smbv1_conn(self):
-        self.logger.debug("Creating smbv1 connection object")
-        try:
-            self.conn = SMBConnection(self.host, self.host, None, 445, preferredDialect=SMB_DIALECT)
-            self.smbv1 = True
-            if self.conn:
-                self.logger.debug("SMBv1 Connection successful")
-        except OSError as e:
-            if str(e).find("Connection reset by peer") != -1:
-                self.logger.debug(f"SMBv1 might be disabled on {self.host}")
-            return False
-        except Exception as e:
-            self.logger.debug(f"Error creating SMBv1 connection to {self.host}: {e}")
-            return False
-        return True
-
-    def create_smbv3_conn(self):
-        self.logger.debug("Creating smbv3 connection object")
-        try:
-            self.conn = SMBConnection(self.host, self.host, None, 445)
-            self.smbv1 = False
-            if self.conn:
-                self.logger.debug("SMBv3 Connection successful")
-        except OSError:
-            return False
-        except Exception as e:
-            self.logger.debug(f"Error creating SMBv3 connection to {self.host}: {e}")
-            return False
-
-        return True
-
-    def create_conn_obj(self):
-        return bool(self.args.no_smb or self.create_smbv1_conn() or self.create_smbv3_conn())
-
     def get_sid(self):
         self.logger.highlight(f"Domain SID {self.sid_domain}")
 

nxc/protocols/ldap/proto_args.py

@@ -5,7 +5,6 @@ def proto_args(parser, parents):
     ldap_parser = parser.add_parser("ldap", help="own stuff using LDAP", parents=parents, formatter_class=DisplayDefaultsNotNone)
     ldap_parser.add_argument("-H", "--hash", metavar="HASH", dest="hash", nargs="+", default=[], help="NTLM hash(es) or file(s) containing NTLM hashes")
     ldap_parser.add_argument("--port", type=int, default=389, help="LDAP port")
-    ldap_parser.add_argument("--no-smb", action="store_true", help="No smb connection")
 
     dgroup = ldap_parser.add_mutually_exclusive_group()
     dgroup.add_argument("-d", metavar="DOMAIN", dest="domain", type=str, default=None, help="domain to authenticate to")