Merge pull request Pennyw0rth#671 from Pennyw0rth/fix_host_info

Author: mpgn

nxc/helpers/misc.py

@@ -4,6 +4,7 @@
 import inspect
 import os
 
+from ipaddress import ip_address
 
 def identify_target_file(target_file):
     with open(target_file) as target_file_handle:
@@ -77,3 +78,10 @@ def _access_check(fn, mode):
                 name = os.path.join(p, thefile)
                 if _access_check(name, mode):
                     return name
+
+def detect_if_ip(target):
+    try:
+        ip_address(target)
+        return True
+    except Exception:
+        return False

nxc/protocols/ldap/resolution.py

@@ -0,0 +1,63 @@
+from re import sub, I
+from errno import EHOSTUNREACH, ETIMEDOUT, ENETUNREACH
+from OpenSSL.SSL import SysCallError
+
+from impacket.ldap import ldap as ldap_impacket
+from impacket.ldap import ldapasn1 as ldapasn1_impacket
+
+from nxc.parsers.ldap_results import parse_result_attributes
+from nxc.logger import nxc_logger
+
+class LDAPResolution:
+
+    def __init__(self, host):
+        self.host = host
+
+    def get_resolution(self):
+        target = ""
+        target_domain = ""
+        base_dn = ""
+        try:
+            ldap_url = f"ldap://{self.host}"
+            nxc_logger.info(f"Connecting to {ldap_url} with no baseDN")
+            try:
+                self.ldap_connection = ldap_impacket.LDAPConnection(ldap_url, dstIp=self.host)
+                if self.ldap_connection:
+                    nxc_logger.debug(f"ldap_connection: {self.ldap_connection}")
+            except SysCallError as e:
+                nxc_logger.fail(f"LDAP connection to {ldap_url} failed: {e}")
+                return False
+
+            resp = self.ldap_connection.search(
+                scope=ldapasn1_impacket.Scope("baseObject"),
+                attributes=["defaultNamingContext", "dnsHostName"],
+                sizeLimit=0,
+            )
+            resp_parsed = parse_result_attributes(resp)[0]
+
+            target = resp_parsed["dnsHostName"]
+            base_dn = resp_parsed["defaultNamingContext"]
+            target_domain = sub(
+                ",DC=",
+                ".",
+                base_dn[base_dn.lower().find("dc="):],
+                flags=I,
+            )[3:]
+            # Extract machine name from target (hostname part of FQDN)
+            if target:
+                machine_name = target.split(".")[0]
+                nxc_logger.debug(f"Extracted machine name: {machine_name}")
+
+            self.ldap_connection.close()
+        except ConnectionRefusedError as e:
+            nxc_logger.debug(f"{e} on host {self.host}")
+            return False
+        except OSError as e:
+            if e.errno in (EHOSTUNREACH, ENETUNREACH, ETIMEDOUT):
+                nxc_logger.info(f"Error connecting to {self.host} - {e}")
+                return False
+            else:
+                nxc_logger.error(f"Error getting ldap info {e}")
+
+        nxc_logger.debug(f"Target: {machine_name}.{target_domain}; target_domain: {target_domain}; base_dn: {base_dn}")
+        return machine_name, target_domain

nxc/protocols/smb.py

@@ -55,6 +55,8 @@
 from nxc.helpers.logger import highlight
 from nxc.helpers.bloodhound import add_user_bh
 from nxc.helpers.powershell import create_ps_command
+from nxc.helpers.misc import detect_if_ip
+from nxc.protocols.ldap.resolution import LDAPResolution
 
 from dploot.triage.vaults import VaultsTriage
 from dploot.triage.browser import BrowserTriage, LoginData, GoogleRefreshToken, Cookie
@@ -125,6 +127,7 @@ def __init__(self, args, db, host):
         self.no_ntlm = False
         self.protocol = "SMB"
         self.is_guest = None
+        self.isdc = False
 
         connection.__init__(self, args, db, host)
 
@@ -185,20 +188,30 @@ def enum_host_info(self):
             if not self.targetDomain:   # Not sure if that can even happen but now we are safe
                 self.targetDomain = self.hostname
         else:
-            # If we can't authenticate with NTLM and the target is supplied as a FQDN we must parse it
             try:
-                import socket
-                socket.inet_aton(self.host)
-                self.logger.debug("NTLM authentication not available! Authentication will fail without a valid hostname and domain name")
-                self.hostname = self.host
-                self.targetDomain = self.host
+                # If we know the host is a DC we can still get the hostname over LDAP if NTLM is not available
+                if self.is_host_dc() and detect_if_ip(self.host):
+                    self.hostname, self.domain = LDAPResolution(self.host).get_resolution()
+                    self.targetDomain = self.domain
+                # If we can't authenticate with NTLM and the target is supplied as a FQDN we must parse it
+                else:
+                    # Check if the host is a valid IP address, if not we parse the FQDN in the Exception
+                    import socket
+                    socket.inet_aton(self.host)
+                    self.logger.debug("NTLM authentication not available! Authentication will fail without a valid hostname and domain name")
+                    self.hostname = self.host
+                    self.targetDomain = self.host
             except OSError:
                 if self.host.count(".") >= 1:
                     self.hostname = self.host.split(".")[0]
                     self.targetDomain = ".".join(self.host.split(".")[1:])
                 else:
                     self.hostname = self.host
                     self.targetDomain = self.host
+            except Exception as e:
+                self.logger.debug(f"Error getting hostname from LDAP: {e}")
+                self.hostname = self.host
+                self.targetDomain = self.host
 
         if self.args.domain:
             self.domain = self.args.domain
@@ -283,21 +296,12 @@ def print_host_info(self):
         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}) {ntlm}")
 
         if self.args.generate_hosts_file or self.args.generate_krb5_file:
-            from impacket.dcerpc.v5 import nrpc, epm
-            self.logger.debug("Performing authentication attempts...")
-            isdc = False
-            try:
-                epm.hept_map(self.host, nrpc.MSRPC_UUID_NRPC, protocol="ncacn_ip_tcp")
-                isdc = True
-            except DCERPCException:
-                self.logger.debug("Error while connecting to host: DCERPCException, which means this is probably not a DC!")
-
             if self.args.generate_hosts_file:
                 with open(self.args.generate_hosts_file, "a+") as host_file:
-                    dc_part = f" {self.targetDomain}" if isdc else ""
+                    dc_part = f" {self.targetDomain}" if self.isdc else ""
                     host_file.write(f"{self.host}     {self.hostname}.{self.targetDomain}{dc_part} {self.hostname}\n")
-                    self.logger.debug(f"{self.host}    {self.hostname}.{self.targetDomain}{dc_part} {self.hostname}")
-            elif self.args.generate_krb5_file and isdc:
+                    self.logger.debug(f"Line added to {self.args.generate_hosts_file} {self.host}    {self.hostname}.{self.targetDomain}{dc_part} {self.hostname}")
+            elif self.args.generate_krb5_file and self.isdc:
                 with open(self.args.generate_krb5_file, "w+") as host_file:
                     data = f"""
 [libdefaults]
@@ -658,6 +662,18 @@ def generate_tgt(self):
         except Exception as e:
             self.logger.fail(f"Failed to get TGT: {e}")
 
+    def is_host_dc(self):
+        from impacket.dcerpc.v5 import nrpc, epm
+        self.logger.debug("Performing authentication attempts...")
+        try:
+            epm.hept_map(self.host, nrpc.MSRPC_UUID_NRPC, protocol="ncacn_ip_tcp")
+            self.isdc = True
+            return True
+        except DCERPCException:
+            self.logger.debug("Error while connecting to host: DCERPCException, which means this is probably not a DC!")
+        self.isdc = False
+        return False
+
     @requires_admin
     def execute(self, payload=None, get_output=False, methods=None) -> str:
         """