Merge pull request #215 from Pennyw0rth/marshall-ldap-users

Update LDAP users lookup to match SMB

nxc/parsers/ldap_results.py

@@ -0,0 +1,13 @@
+from impacket.ldap import ldapasn1 as ldapasn1_impacket
+
+def parse_result_attributes(ldap_response):
+    parsed_response = []
+    for entry in ldap_response:
+        # SearchResultReferences may be returned
+        if not isinstance(entry, ldapasn1_impacket.SearchResultEntry):
+            continue
+        attribute_map = {}
+        for attribute in entry["attributes"]:
+            attribute_map[str(attribute["type"])] = str(attribute["vals"][0])
+        parsed_response.append(attribute_map)
+    return parsed_response

nxc/protocols/ldap.py

@@ -5,7 +5,7 @@ import hmac
 import os
 import socket
 from binascii import hexlify
-from datetime import datetime
+from datetime import datetime, timedelta
 from re import sub, I
 from zipfile import ZipFile
 from termcolor import colored
@@ -38,6 +38,7 @@ from nxc.logger import NXCAdapter, nxc_logger
 from nxc.protocols.ldap.bloodhound import BloodHound
 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
 
 ldap_error_status = {
     "1": "STATUS_NOT_SUPPORTED",
@@ -372,7 +373,6 @@ class ldap(connection):
             used_ccache = " from ccache" if useCache else f":{process_secret(kerb_pass)}"
             out = f"{domain}\\{self.username}{used_ccache} {self.mark_pwned()}"
 
-
             self.logger.extra["protocol"] = "LDAP"
             self.logger.extra["port"] = "636" if (self.args.gmsa or self.port == 636) else "389"
             self.logger.success(out)
@@ -753,38 +753,52 @@ class ldap(connection):
         return False
 
     def users(self):
-        # Building the search filter
-        search_filter = "(sAMAccountType=805306368)" if self.username != "" else "(objectclass=*)"
-        attributes = [
-            "sAMAccountName",
-            "description",
-            "badPasswordTime",
-            "badPwdCount",
-            "pwdLastSet",
-        ]
+        """
+        Retrieves user information from the LDAP server.
+
+        Args:
+        ----
+            input_attributes (list): Optional. List of attributes to retrieve for each user.
+
+        Returns:
+        -------
+            None
+        """
+        if len(self.args.users) > 0:
+            self.logger.debug(f"Dumping users: {', '.join(self.args.users)}")
+            search_filter = f"(|{''.join(f'(sAMAccountName={user})' for user in self.args.users)})"
+        else:
+            self.logger.debug("Trying to dump all users")
+            search_filter = "(sAMAccountType=805306368)" if self.username != "" else "(objectclass=*)"
+
+        # default to these attributes to mirror the SMB --users functionality
+        request_attributes = ["sAMAccountName", "description", "badPwdCount", "pwdLastSet"]
+        resp = self.search(search_filter, request_attributes, sizeLimit=0)
 
-        resp = self.search(search_filter, attributes, sizeLimit=0)
         if resp:
-            self.logger.display(f"Total of records returned {len(resp):d}")
+            # I think this was here for anonymous ldap bindings, so I kept it, but we might just want to remove it
+            if self.username == "":
+                self.logger.display(f"Total records returned: {len(resp):d}")
                 for item in resp:
                     if isinstance(item, ldapasn1_impacket.SearchResultEntry) is not True:
                         continue
-                sAMAccountName = ""
-                description = ""
-                try:
-                    if self.username == "":
                     self.logger.highlight(f"{item['objectName']}")
-                    else:
-                        for attribute in item["attributes"]:
-                            if str(attribute["type"]) == "sAMAccountName":
-                                sAMAccountName = str(attribute["vals"][0])
-                            elif str(attribute["type"]) == "description":
-                                description = str(attribute["vals"][0])
-                        self.logger.highlight(f"{sAMAccountName:<30} {description}")
-                except Exception as e:
-                    self.logger.debug(f"Skipping item, cannot process due to error {e}")
                 return
 
+            users = parse_result_attributes(resp)
+            # we print the total records after we parse the results since often SearchResultReferences are returned
+            self.logger.display(f"Total records returned: {len(users):d}")
+            self.logger.highlight(f"{'-Username-':<30}{'-Last PW Set-':<20}{'-BadPW-':<8}{'-Description-':<60}")
+            for user in users:
+                # TODO: functionize this - we do this calculation in a bunch of places, different, including in the `pso` module
+                timestamp_seconds = int(user.get("pwdLastSet", "")) / 10**7
+                start_date = datetime(1601, 1, 1)
+                parsed_pw_last_set = (start_date + timedelta(seconds=timestamp_seconds)).replace(microsecond=0).strftime("%Y-%m-%d %H:%M:%S")
+                if parsed_pw_last_set == "1601-01-01 00:00:00":
+                    parsed_pw_last_set = "<never>"
+                # we default attributes to blank strings if they don't exist in the dict
+                self.logger.highlight(f"{user.get('sAMAccountName', ''):<30}{parsed_pw_last_set:<20}{user.get('badPwdCount', ''):<8}{user.get('description', ''):<60}")
+
     def groups(self):
         # Building the search filter
         search_filter = "(objectCategory=group)"

nxc/protocols/ldap/proto_args.py

@@ -20,7 +20,7 @@ def proto_args(parser, std_parser, module_parser):
     vgroup.add_argument("--trusted-for-delegation", action="store_true", help="Get the list of users and computers with flag TRUSTED_FOR_DELEGATION")
     vgroup.add_argument("--password-not-required", action="store_true", help="Get the list of users with flag PASSWD_NOTREQD")
     vgroup.add_argument("--admin-count", action="store_true", help="Get objets that had the value adminCount=1")
-    vgroup.add_argument("--users", action="store_true", help="Enumerate enabled domain users")
+    vgroup.add_argument("--users", nargs="*", help="Enumerate enabled domain users")
     vgroup.add_argument("--groups", action="store_true", help="Enumerate domain groups")
     vgroup.add_argument("--dc-list", action="store_true", help="Enumerate Domain Controllers")
     vgroup.add_argument("--get-sid", action="store_true", help="Get domain sid")

nxc/protocols/smb.py

@@ -1000,10 +1000,8 @@ class smb(connection):
         return groups
 
     def users(self):
-        if len(self.args.users) > 1:
-            self.logger.display(f"Dumping users: {', '.join(self.args.users)}")
-        else:
-            self.logger.info("Trying to dump local users with SAMRPC protocol") 
+        if len(self.args.users) > 0:
+            self.logger.debug(f"Dumping users: {', '.join(self.args.users)}")
              
         return UserSamrDump(self).dump(self.args.users)