Add LDAP query obfuscation support via ldapx

nxc/protocols/ldap.py

@@ -50,6 +50,8 @@
 from nxc.helpers.negotiate_parser import parse_challenge
 from nxc.paths import CONFIG_PATH
 
+import ldapx
+
 ldap_error_status = {
     "1": "STATUS_NOT_SUPPORTED",
     "533": "STATUS_ACCOUNT_DISABLED",
@@ -89,6 +91,7 @@ def __init__(self, args, db, host):
         self.sid_domain = ""
         self.scope = None
         self.configuration_context = ""
+        self.obfuscate = args.obfuscate
 
         connection.__init__(self, args, db, host)
 
@@ -103,6 +106,9 @@ def proto_logger(self):
         )
 
     def create_conn_obj(self):
+        if self.obfuscate:
+            self.logger.info("LDAP query obfuscation enabled")
+
         try:
             proto = "ldaps" if self.port == 636 else "ldap"
             ldap_url = f"{proto}://{self.host}"
@@ -660,6 +666,18 @@ def search(self, searchFilter, attributes, sizeLimit=0, baseDN=None, searchContr
         elif baseDN is None:
             baseDN = self.baseDN
 
+        if self.obfuscate:
+            try:
+                if searchFilter:
+                    searchFilter = ldapx.obfuscate_filter(searchFilter, "CSG")
+                if baseDN:
+                    baseDN = ldapx.obfuscate_basedn(baseDN, "CX")
+                if attributes:
+                    attributes = ldapx.obfuscate_attrlist(attributes, "CR")
+                self.logger.debug(f"Obfuscated baseDN: {baseDN}")
+            except Exception as e:
+                self.logger.debug(f"ldapx obfuscation failed, using original query: {e}")
+
         try:
             if self.ldap_connection:
                 self.logger.debug(f"Search Filter={searchFilter}")
@@ -1295,12 +1313,7 @@ def gmsa(self):
                     sids = [ace["Ace"]["Sid"].formatCanonical() for ace in dacl["Dacl"]["Data"] if ace["AceType"] == 0x00]
                     self.logger.debug(f"msDS-GroupMSAMembership: {sids}")
                     search_filter = "(|" + "".join([f"(objectSid={sid})" for sid in sids]) + ")"
-                    resp = self.ldap_connection.search(
-                        searchBase=self.baseDN,
-                        searchFilter=search_filter,
-                        attributes=["sAMAccountName"],
-                        sizeLimit=0,
-                    )
+                    resp = self.search(search_filter, ["sAMAccountName"])
                     resp_parsed = parse_result_attributes(resp)
                     if len(resp_parsed) > 1:
                         principal_with_read = [f"{item['sAMAccountName']}" for item in resp_parsed]
@@ -1340,12 +1353,7 @@ def gmsa_convert_id(self):
             else:
                 # getting the gmsa account
                 search_filter = "(objectClass=msDS-GroupManagedServiceAccount)"
-                gmsa_accounts = self.ldap_connection.search(
-                    searchBase=self.baseDN,
-                    searchFilter=search_filter,
-                    attributes=["sAMAccountName"],
-                    sizeLimit=0,
-                )
+                gmsa_accounts = self.search(search_filter, ["sAMAccountName"])
                 gmsa_accounts_parsed = parse_result_attributes(gmsa_accounts)
                 if gmsa_accounts_parsed:
                     self.logger.debug(f"Total of records returned {len(gmsa_accounts_parsed):d}")
@@ -1363,12 +1371,7 @@ def gmsa_decrypt_lsa(self):
                 gmsa_id, gmsa_pass = self.args.gmsa_decrypt_lsa.split("_")[4].split(":")
                 # getting the gmsa account
                 search_filter = "(objectClass=msDS-GroupManagedServiceAccount)"
-                gmsa_accounts = self.ldap_connection.search(
-                    searchBase=self.baseDN,
-                    searchFilter=search_filter,
-                    attributes=["sAMAccountName"],
-                    sizeLimit=0,
-                )
+                gmsa_accounts = self.search(search_filter, ["sAMAccountName"])
                 gmsa_accounts_parsed = parse_result_attributes(gmsa_accounts)
                 if gmsa_accounts_parsed:
                     self.logger.debug(f"Total of records returned {len(gmsa_accounts):d}")

nxc/protocols/ldap/proto_args.py

@@ -45,4 +45,6 @@ def proto_args(parser, parents):
     bgroup.add_argument("--bloodhound", action="store_true", help="Perform a Bloodhound scan")
     bgroup.add_argument("-c", "--collection", default="Default", help="Which information to collect. Supported: Group, LocalAdmin, Session, Trusts, Default, DCOnly, DCOM, RDP, PSRemote, LoggedOn, Container, ObjectProps, ACL, ADCS, All. You can specify more than one by separating them with a comma.")
 
+    ldap_parser.add_argument("--obfuscate", action="store_true", help="Enable LDAP query obfuscation via ldapx")
+
     return parser

pyproject.toml

@@ -26,6 +26,7 @@ dependencies = [
     "dploot>=3.2.2",
     "dsinternals>=1.2.4",
     "jwt>=1.3.1",
+    "ldapx>=0.4.0",
     "lsassy>=3.1.11",
     "masky>=0.2.1",
     "minikerberos>=0.4.1",