Merge pull request Pennyw0rth#811 from Pennyw0rth/neff-fix-daclread

Add ldap parsing to daclread

Author: NeffIsBack

nxc/modules/daclread.py

@@ -1,16 +1,17 @@
 import binascii
 import codecs
 import json
-import re
 import datetime
 from enum import Enum
 from impacket.ldap import ldaptypes
 from impacket.uuid import bin_to_string
 from nxc.helpers.msada_guids import SCHEMA_OBJECTS, EXTENDED_RIGHTS
-from ldap3.protocol.formatters.formatters import format_sid
+from nxc.parsers.ldap_results import parse_result_attributes
 from ldap3.utils.conv import escape_filter_chars
 from ldap3.protocol.microsoft import security_descriptor_control
 import sys
+import traceback
+from os.path import isfile
 
 OBJECT_TYPES_GUID = {}
 OBJECT_TYPES_GUID.update(SCHEMA_OBJECTS)
@@ -227,12 +228,12 @@ def options(self, context, module_options):
 
         if module_options and "TARGET" in module_options:
             context.log.debug("There is a target specified!")
-            if re.search(r"^(.+)\/([^\/]+)$", module_options["TARGET"]) is not None:
+            if isfile(module_options["TARGET"]):
                 try:
                     self.target_file = open(module_options["TARGET"])  # noqa: SIM115
                     self.target_sAMAccountName = None
                 except Exception:
-                    context.log.fail("The file doesn't exist or cannot be openned.")
+                    context.log.fail("The file doesn't exist or cannot be opened.")
             else:
                 context.log.debug(f"Setting target_sAMAccountName to {module_options['TARGET']}")
                 self.target_sAMAccountName = module_options["TARGET"]
@@ -274,37 +275,50 @@ def on_login(self, context, connection):
         context.log.highlight("Be careful, this module cannot read the DACLS recursively.")
         self.baseDN = connection.ldap_connection._baseDN
         self.ldap_session = connection.ldap_connection
+        self.connection = connection
+        self.context = context
 
         # Searching for the principal SID
         if self.principal_sAMAccountName is not None:
-            _lookedup_principal = self.principal_sAMAccountName
             try:
-                self.principal_sid = format_sid(
-                    self.ldap_session.search(
-                        searchBase=self.baseDN,
-                        searchFilter=f"(sAMAccountName={escape_filter_chars(_lookedup_principal)})",
-                        attributes=["objectSid"],
-                    )[0][1][0][1][0]
+                resp = connection.search(
+                    searchFilter=f"(sAMAccountName={escape_filter_chars(self.principal_sAMAccountName)})",
+                    attributes=["objectSid"],
                 )
+                resp_parsed = parse_result_attributes(resp)[0]
+                self.principal_sid = resp_parsed["objectSid"]
                 context.log.highlight(f"Found principal SID to filter on: {self.principal_sid}")
-            except Exception:
-                context.log.fail(f"Principal SID not found in LDAP ({_lookedup_principal})")
-                sys.exit(1)
+            except Exception as e:
+                context.log.fail(f"Principal SID not found in LDAP ({self.principal_sAMAccountName})")
+                context.log.debug(f"Exception: {e}, {traceback.format_exc()}")
+                return
 
         # Searching for the targets SID and their Security Descriptors
         # If there is only one target
         if (self.target_sAMAccountName or self.target_DN) and self.target_file is None:
-            # Searching for target account with its security descriptor
             try:
-                self.search_target_principal_security_descriptor(context, connection)
+                # Searching for target account with its security descriptor
+                if self.target_sAMAccountName:  # noqa: SIM108
+                    search_filter = f"(sAMAccountName={escape_filter_chars(self.target_sAMAccountName)})"
+                else:
+                    search_filter = f"(distinguishedName={escape_filter_chars(self.target_DN)})"
+
+                resp = connection.search(
+                    searchFilter=search_filter,
+                    attributes=["distinguishedName", "nTSecurityDescriptor"],
+                    searchControls=security_descriptor_control(sdflags=0x04),
+                )
+                resp_parsed = parse_result_attributes(resp)[0]
+
                 # Extract security descriptor data
-                self.target_principal_dn = self.target_principal[0]
-                self.principal_raw_security_descriptor = str(self.target_principal[1][0][1][0]).encode("latin-1")
+                self.target_principal_dn = resp_parsed["distinguishedName"]
+                self.principal_raw_security_descriptor = resp_parsed["nTSecurityDescriptor"]
                 self.principal_security_descriptor = ldaptypes.SR_SECURITY_DESCRIPTOR(data=self.principal_raw_security_descriptor)
-                context.log.highlight(f"Target principal found in LDAP ({self.target_principal[0]})")
-            except Exception:
+                context.log.highlight(f"Target principal found in LDAP ({self.target_principal_dn})")
+            except Exception as e:
                 context.log.fail(f"Target SID not found in LDAP ({self.target_sAMAccountName})")
-                sys.exit(1)
+                context.log.debug(f"Exception: {e}, {traceback.format_exc()}")
+                return
 
             if self.action == "read":
                 self.read(context)
@@ -318,10 +332,16 @@ def on_login(self, context, connection):
                 try:
                     self.target_sAMAccountName = target.strip()
                     # Searching for target account with its security descriptor
-                    self.search_target_principal_security_descriptor(context, connection)
+                    resp = connection.search(
+                        searchFilter=f"(sAMAccountName={escape_filter_chars(self.target_sAMAccountName)})",
+                        attributes=["distinguishedName", "nTSecurityDescriptor"],
+                        searchControls=security_descriptor_control(sdflags=0x04),
+                    )
+                    resp_parsed = parse_result_attributes(resp)[0]
+
                     # Extract security descriptor data
-                    self.target_principal_dn = self.target_principal[0]
-                    self.principal_raw_security_descriptor = str(self.target_principal[1][0][1][0]).encode("latin-1")
+                    self.target_principal_dn = resp_parsed["distinguishedName"]
+                    self.principal_raw_security_descriptor = resp_parsed["nTSecurityDescriptor"]
                     self.principal_security_descriptor = ldaptypes.SR_SECURITY_DESCRIPTOR(data=self.principal_raw_security_descriptor)
                     context.log.highlight(f"Target principal found in LDAP ({self.target_sAMAccountName})")
                 except Exception:
@@ -355,72 +375,22 @@ def backup(self, context):
         context.log.highlight("DACL backed up to %s", self.filename)
         self.filename = None
 
-    # Attempts to retrieve the DACL in the Security Descriptor of the specified target
-    def search_target_principal_security_descriptor(self, context, connection):
-        _lookedup_principal = ""
-        # Set SD flags to only query for DACL
-        controls = security_descriptor_control(sdflags=0x04)
-        if self.target_sAMAccountName is not None:
-            _lookedup_principal = self.target_sAMAccountName
-            target = self.ldap_session.search(
-                searchBase=self.baseDN,
-                searchFilter=f"(sAMAccountName={escape_filter_chars(_lookedup_principal)})",
-                attributes=["nTSecurityDescriptor"],
-                searchControls=controls,
-            )
-        if self.target_DN is not None:
-            _lookedup_principal = self.target_DN
-            target = self.ldap_session.search(
-                searchBase=_lookedup_principal,
-                searchFilter=f"(distinguishedName={_lookedup_principal})",
-                attributes=["nTSecurityDescriptor"],
-                searchControls=controls,
-            )
-        try:
-            self.target_principal = target[0]
-        except Exception:
-            context.log.fail(f"Principal not found in LDAP ({_lookedup_principal}), probably an LDAP session issue.")
-            sys.exit(0)
-
-    # Attempts to retrieve the SID and Distinguisehd Name from the sAMAccountName
-    # Not used for the moment
-    #   - samname : a sAMAccountName
-    def get_user_info(self, context, samname):
-        self.ldap_session.search(
-            searchBase=self.baseDN,
-            searchFilter=f"(sAMAccountName={escape_filter_chars(samname)})",
-            attributes=["objectSid"],
-        )
-        try:
-            dn = self.ldap_session.entries[0].entry_dn
-            sid = format_sid(self.ldap_session.entries[0]["objectSid"].raw_values[0])
-            return dn, sid
-        except Exception:
-            context.log.fail(f"User not found in LDAP: {samname}")
-            return False
-
-    # Attempts to resolve a SID and return the corresponding samaccountname
-    #   - sid : the SID to resolve
-    def resolveSID(self, context, sid):
+    def resolveSID(self, sid):
+        """Resolves a SID to its corresponding sAMAccountName."""
         # Tries to resolve the SID from the well known SIDs
         if sid in WELL_KNOWN_SIDS:
             return WELL_KNOWN_SIDS[sid]
+
         # Tries to resolve the SID from the LDAP domain dump
-        else:
-            try:
-                self.ldap_session.search(
-                    searchBase=self.baseDN,
-                    searchFilter=f"(objectSid={sid})",
-                    attributes=["sAMAccountName"],
-                )[0][0]
-                return self.ldap_session.search(
-                    searchBase=self.baseDN,
-                    searchFilter=f"(objectSid={sid})",
-                    attributes=["sAMAccountName"],
-                )[0][1][0][1][0]
-            except Exception:
-                context.log.debug(f"SID not found in LDAP: {sid}")
-                return ""
+        try:
+            resp = self.connection.search(
+                searchFilter=f"(objectSid={sid})",
+                attributes=["sAMAccountName"],
+            )
+            return parse_result_attributes(resp)[0]["sAMAccountName"]
+        except Exception:
+            self.context.log.debug(f"SID not found in LDAP: {sid}")
+            return ""
 
     # Parses a full DACL
     #   - dacl : the DACL to parse, submitted in a Security Desciptor format
@@ -458,7 +428,7 @@ def parse_ace(self, context, ace):
             # Extracts the access mask (by parsing the simple permissions) and the principal's SID
             if ace["TypeName"] in ["ACCESS_ALLOWED_ACE", "ACCESS_DENIED_ACE"]:
                 access_mask = f"{', '.join(self.parse_perms(ace['Ace']['Mask']['Mask']))} (0x{ace['Ace']['Mask']['Mask']:x})"
-                trustee_sid = f"{self.resolveSID(context, ace['Ace']['Sid'].formatCanonical()) or 'UNKNOWN'} ({ace['Ace']['Sid'].formatCanonical()})"
+                trustee_sid = f"{self.resolveSID(ace['Ace']['Sid'].formatCanonical()) or 'UNKNOWN'} ({ace['Ace']['Sid'].formatCanonical()})"
                 parsed_ace = {
                     "Access mask": access_mask,
                     "Trustee (SID)": trustee_sid
@@ -486,7 +456,7 @@ def parse_ace(self, context, ace):
                         parsed_ace["Inherited type (GUID)"] = f"UNKNOWN ({inh_obj_type})"
                 # Extract the Trustee SID (the object that has the right over the DACL bearer)
                 parsed_ace["Trustee (SID)"] = "{} ({})".format(
-                    self.resolveSID(context, ace["Ace"]["Sid"].formatCanonical()) or "UNKNOWN",
+                    self.resolveSID(ace["Ace"]["Sid"].formatCanonical()) or "UNKNOWN",
                     ace["Ace"]["Sid"].formatCanonical(),
                 )
         else:  # if the ACE is not an access allowed
@@ -501,7 +471,7 @@ def parse_ace(self, context, ace):
 
     def print_parsed_dacl(self, context, parsed_dacl):
         """Prints a full DACL by printing each parsed ACE
-        
+
         parsed_dacl : a parsed DACL from parse_dacl()
         """
         context.log.debug("Printing parsed DACL")

nxc/protocols/ldap.py

@@ -694,7 +694,7 @@ def getUnixTime(self, t):
         t /= 10000000
         return t
 
-    def search(self, searchFilter, attributes, sizeLimit=0, baseDN=None) -> list:
+    def search(self, searchFilter, attributes, sizeLimit=0, baseDN=None, searchControls=None) -> list:
         if baseDN is None and self.args.base_dn is not None:
             baseDN = self.args.base_dn
         elif baseDN is None:
@@ -712,7 +712,7 @@ def search(self, searchFilter, attributes, sizeLimit=0, baseDN=None) -> list:
                     searchFilter=searchFilter,
                     attributes=attributes,
                     sizeLimit=sizeLimit,
-                    searchControls=paged_search_control,
+                    searchControls=searchControls if searchControls else paged_search_control,
                 )
         except ldap_impacket.LDAPSearchError as e:
             if "sizeLimitExceeded" in str(e):