Merge pull request Pennyw0rth#832 from tiagomanunes/improve-daclread-targetdn-from-file

Improve daclread: also allow passing a file for TARGET_DN, and refactor

Author: NeffIsBack

nxc/modules/daclread.py

@@ -1,5 +1,4 @@
 import binascii
-import codecs
 import json
 import datetime
 from enum import Enum
@@ -189,6 +188,12 @@ class ALLOWED_OBJECT_ACE_MASK_FLAGS(Enum):
     Self = ldaptypes.ACCESS_ALLOWED_OBJECT_ACE.ADS_RIGHT_DS_SELF
 
 
+SEARCH_FILTERS = {
+    "TARGET": lambda target: f"(sAMAccountName={escape_filter_chars(target)})",
+    "TARGET_DN": lambda target: f"(distinguishedName={escape_filter_chars(target)})"
+}
+
+
 class NXCModule:
     """Module to read and backup the Discretionary Access Control List of one or multiple objects.
 
@@ -205,6 +210,14 @@ def __init__(self, context=None, module_options=None):
         self.context = context
         self.module_options = module_options
 
+        # Initialize module variables
+        self.principal_sAMAccountName = None
+        self.principal_sid = None
+        self.action = "read"
+        self.ace_type = "allowed"
+        self.rights = None
+        self.rights_guid = None
+
     def options(self, context, module_options):
         """
         Be careful, this module cannot read the DACLS recursively. 
@@ -226,57 +239,48 @@ def options(self, context, module_options):
             context.log.fail("Select an option, example: -M daclread -o TARGET=Administrator ACTION=read")
             sys.exit(1)
 
-        if module_options and "TARGET" in module_options:
-            context.log.debug("There is a target specified!")
-            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 opened.")
-            else:
-                context.log.debug(f"Setting target_sAMAccountName to {module_options['TARGET']}")
-                self.target_sAMAccountName = module_options["TARGET"]
-                self.target_file = None
-            self.target_DN = None
+        self.targets = []
         self.target_SID = None
-        if module_options and "TARGET_DN" in module_options:
-            self.target_DN = module_options["TARGET_DN"]
-            self.target_sAMAccountName = None
-            self.target_file = None
 
-        if module_options and "PRINCIPAL" in module_options:
+        for option in "TARGET", "TARGET_DN":
+            if option in module_options:
+                context.log.debug("There is a target specified!")
+                if isfile(module_options[option]):
+                    try:
+                        target_file = open(module_options[option])  # noqa: SIM115
+                        for line in target_file:
+                            context.log.debug(f"Adding target from file: {line}")
+                            self.targets.append((line.strip(), SEARCH_FILTERS[option]))
+                    except Exception:
+                        context.log.fail("The file doesn't exist or cannot be opened.")
+                else:
+                    context.log.debug(f"Adding target: {module_options[option]}")
+                    self.targets.append((module_options[option].strip(), SEARCH_FILTERS[option]))
+
+        if not self.targets:
+            context.log.fail("No target specified, please specify at least one target with the TARGET or TARGET_DN options.")
+            sys.exit(1)
+
+        if "PRINCIPAL" in module_options:
             self.principal_sAMAccountName = module_options["PRINCIPAL"]
-        else:
-            self.principal_sAMAccountName = None
-        self.principal_sid = None
 
-        if module_options and "ACTION" in module_options:
+        if "ACTION" in module_options:
             self.action = module_options["ACTION"]
-        else:
-            self.action = "read"
-        if module_options and "ACE_TYPE" in module_options:
+
+        if "ACE_TYPE" in module_options:
             self.ace_type = module_options["ACE_TYPE"]
-        else:
-            self.ace_type = "allowed"
-        if module_options and "RIGHTS" in module_options:
+            
+        if "RIGHTS" in module_options:
             self.rights = module_options["RIGHTS"]
-        else:
-            self.rights = None
-        if module_options and "RIGHTS_GUID" in module_options:
+
+        if "RIGHTS_GUID" in module_options:
             self.rights_guid = module_options["RIGHTS_GUID"]
-        else:
-            self.rights_guid = None
-        self.filename = None
 
     def on_login(self, context, connection):
-        self.context = context
         """On a successful LDAP login we perform a search for the targets' SID, their Security Descriptors and the principal's SID if there is one specified"""
         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
+        self.connection = connection
 
         # Searching for the principal SID
         if self.principal_sAMAccountName is not None:
@@ -294,86 +298,49 @@ def on_login(self, context, connection):
                 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:
+        for target, search_filter in self.targets:
             try:
                 # 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,
+                    searchFilter=search_filter(target),
                     attributes=["distinguishedName", "nTSecurityDescriptor"],
                     searchControls=security_descriptor_control(sdflags=0x04),
                 )
                 resp_parsed = parse_result_attributes(resp)[0]
 
                 # Extract security descriptor data
-                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_dn})")
+                target_principal_dn = resp_parsed["distinguishedName"]
+                principal_raw_security_descriptor = resp_parsed["nTSecurityDescriptor"]
+                principal_security_descriptor = ldaptypes.SR_SECURITY_DESCRIPTOR(data=principal_raw_security_descriptor)
+                context.log.highlight(f"Target principal found in LDAP ({target_principal_dn})")
             except Exception as e:
-                context.log.fail(f"Target SID not found in LDAP ({self.target_sAMAccountName})")
+                context.log.fail(f"Target SID not found in LDAP ({target})")
                 context.log.debug(f"Exception: {e}, {traceback.format_exc()}")
-                return
+                continue
 
             if self.action == "read":
-                self.read(context)
+                self.read(principal_security_descriptor)
             if self.action == "backup":
-                self.backup(context)
-
-        # If there are multiple targets
-        else:
-            targets = self.target_file.readlines()
-            for target in targets:
-                try:
-                    self.target_sAMAccountName = target.strip()
-                    # Searching for target account with its security descriptor
-                    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 = 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:
-                    context.log.fail(f"Target SID not found in LDAP ({self.target_sAMAccountName})")
-                    continue
-
-                if self.action == "read":
-                    self.read(context)
-                if self.action == "backup":
-                    self.backup(context)
+                self.backup(target, target_principal_dn, principal_raw_security_descriptor)
 
     # Main read funtion
     # Prints the parsed DACL
-    def read(self, context):
-        parsed_dacl = self.parse_dacl(context, self.principal_security_descriptor["Dacl"])
-        self.print_parsed_dacl(context, parsed_dacl)
+    def read(self, principal_security_descriptor):
+        parsed_dacl = self.parse_dacl(principal_security_descriptor["Dacl"])
+        self.print_parsed_dacl(parsed_dacl)
 
     # Permits to export the DACL of the targets
     # This function is called before any writing action (write, remove or restore)
-    def backup(self, context):
+    def backup(self, target, target_principal_dn, principal_raw_security_descriptor):
         backup = {}
-        backup["sd"] = binascii.hexlify(self.principal_raw_security_descriptor).decode("latin-1")
-        backup["dn"] = str(self.target_principal_dn)
-        if not self.filename:
-            self.filename = "dacledit-{}-{}.bak".format(
-                datetime.datetime.now().strftime("%Y%m%d-%H%M%S"),
-                self.target_sAMAccountName,
-            )
-        with codecs.open(self.filename, "w", "latin-1") as outfile:
+        backup["sd"] = binascii.hexlify(principal_raw_security_descriptor).decode("latin-1")
+        backup["dn"] = str(target_principal_dn)
+
+        timestamp = datetime.datetime.now().strftime("%Y%m%d-%H%M%S")
+        filename = f"dacledit-{timestamp}-{target}.bak"
+        with open(filename, "w", encoding="latin-1") as outfile:
             json.dump(backup, outfile)
-        context.log.highlight("DACL backed up to %s", self.filename)
-        self.filename = None
+        self.context.log.highlight(f"DACL backed up to {filename}")
 
     def resolveSID(self, sid):
         """Resolves a SID to its corresponding sAMAccountName."""
@@ -394,11 +361,11 @@ def resolveSID(self, sid):
 
     # Parses a full DACL
     #   - dacl : the DACL to parse, submitted in a Security Desciptor format
-    def parse_dacl(self, context, dacl):
+    def parse_dacl(self, dacl):
         parsed_dacl = []
-        context.log.debug("Parsing DACL")
+        self.context.log.debug("Parsing DACL")
         for ace in dacl["Data"]:
-            parsed_ace = self.parse_ace(context, ace)
+            parsed_ace = self.parse_ace(ace)
             parsed_dacl.append(parsed_ace)
         return parsed_dacl
 
@@ -413,7 +380,7 @@ def parse_perms(self, access_mask):
 
     # Parses a specified ACE and extract the different values (Flags, Access Mask, Trustee, ObjectType, InheritedObjectType)
     #   - ace : the ACE to parse
-    def parse_ace(self, context, ace):
+    def parse_ace(self, ace):
         # For the moment, only the Allowed and Denied Access ACE are supported
         if ace["TypeName"] in [
             "ACCESS_ALLOWED_ACE",
@@ -455,12 +422,9 @@ def parse_ace(self, context, ace):
                     except KeyError:
                         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(ace["Ace"]["Sid"].formatCanonical()) or "UNKNOWN",
-                    ace["Ace"]["Sid"].formatCanonical(),
-                )
+                parsed_ace["Trustee (SID)"] = f"{self.resolveSID(ace['Ace']['Sid'].formatCanonical()) or 'UNKNOWN'} ({ace['Ace']['Sid'].formatCanonical()})"
         else:  # if the ACE is not an access allowed
-            context.log.debug(f"ACE Type ({ace['TypeName']}) unsupported for parsing yet, feel free to contribute")
+            self.context.log.debug(f"ACE Type ({ace['TypeName']}) unsupported for parsing yet, feel free to contribute")
             _ace_flags = [FLAG.name for FLAG in ACE_FLAGS if ace.hasFlag(FLAG.value)]
             parsed_ace = {
                 "ACE type": ace["TypeName"],
@@ -469,18 +433,18 @@ def parse_ace(self, context, ace):
             }
         return parsed_ace
 
-    def print_parsed_dacl(self, context, parsed_dacl):
+    def print_parsed_dacl(self, 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")
+        self.context.log.debug("Printing parsed DACL")
         # If a specific right or a specific GUID has been specified, only the ACE with this right will be printed
         # If an ACE type has been specified, only the ACE with this type will be specified
         # If a principal has been specified, only the ACE where he is the trustee will be printed
         for i, parsed_ace in enumerate(parsed_dacl):
             print_ace = True
-            context.log.debug(f"{parsed_ace=}, {self.rights=}, {self.rights_guid=}, {self.ace_type=}, {self.principal_sid=}")
+            self.context.log.debug(f"{parsed_ace=}, {self.rights=}, {self.rights_guid=}, {self.ace_type=}, {self.principal_sid=}")
 
             # Filter on specific rights
             if self.rights is not None:
@@ -494,37 +458,37 @@ def print_parsed_dacl(self, context, parsed_dacl):
                     if (self.rights == "ResetPassword") and (("Object type (GUID)" not in parsed_ace) or (RIGHTS_GUID.ResetPassword.value not in parsed_ace["Object type (GUID)"])):
                         print_ace = False
                 except Exception as e:
-                    context.log.debug(f"Error filtering with {parsed_ace=} and {self.rights=}, probably because of ACE type unsupported for parsing yet ({e})")
+                    self.context.log.debug(f"Error filtering with {parsed_ace=} and {self.rights=}, probably because of ACE type unsupported for parsing yet ({e})")
 
             # Filter on specific right GUID
             if self.rights_guid is not None:
                 try:
                     if ("Object type (GUID)" not in parsed_ace) or (self.rights_guid not in parsed_ace["Object type (GUID)"]):
                         print_ace = False
                 except Exception as e:
-                    context.log.debug(f"Error filtering with {parsed_ace=} and {self.rights_guid=}, probably because of ACE type unsupported for parsing yet ({e})")
+                    self.context.log.debug(f"Error filtering with {parsed_ace=} and {self.rights_guid=}, probably because of ACE type unsupported for parsing yet ({e})")
 
             # Filter on ACE type
             if self.ace_type == "allowed":
                 try:
                     if ("ACCESS_ALLOWED_OBJECT_ACE" not in parsed_ace["ACE Type"]) and ("ACCESS_ALLOWED_ACE" not in parsed_ace["ACE Type"]):
                         print_ace = False
                 except Exception as e:
-                    context.log.debug(f"Error filtering with {parsed_ace=} and {self.ace_type=}, probably because of ACE type unsupported for parsing yet ({e})")
+                    self.context.log.debug(f"Error filtering with {parsed_ace=} and {self.ace_type=}, probably because of ACE type unsupported for parsing yet ({e})")
             else:
                 try:
                     if ("ACCESS_DENIED_OBJECT_ACE" not in parsed_ace["ACE Type"]) and ("ACCESS_DENIED_ACE" not in parsed_ace["ACE Type"]):
                         print_ace = False
                 except Exception as e:
-                    context.log.debug(f"Error filtering with {parsed_ace=} and {self.ace_type=}, probably because of ACE type unsupported for parsing yet ({e})")
+                    self.context.log.debug(f"Error filtering with {parsed_ace=} and {self.ace_type=}, probably because of ACE type unsupported for parsing yet ({e})")
 
             # Filter on trusted principal
             if self.principal_sid is not None:
                 try:
                     if self.principal_sid not in parsed_ace["Trustee (SID)"]:
                         print_ace = False
                 except Exception as e:
-                    context.log.debug(f"Error filtering with {parsed_ace=} and {self.principal_sid=}, probably because of ACE type unsupported for parsing yet ({e})")
+                    self.context.log.debug(f"Error filtering with {parsed_ace=} and {self.principal_sid=}, probably because of ACE type unsupported for parsing yet ({e})")
             if print_ace:
                 self.context.log.highlight(f"ACE[{i}] info")
                 self.print_parsed_ace(parsed_ace)