From 29de0ccf349eb256796d183bcdfc1ea4a15dec8d Mon Sep 17 00:00:00 2001 From: termanix <50464194+termanix@users.noreply.github.com> Date: Fri, 18 Oct 2024 16:26:16 +0300 Subject: [PATCH] Used parse_result_attributes for parsing Signed-off-by: termanix <50464194+termanix@users.noreply.github.com> --- nxc/protocols/ldap.py | 104 ++++++++++++++++++++++-------------------- 1 file changed, 55 insertions(+), 49 deletions(-) diff --git a/nxc/protocols/ldap.py b/nxc/protocols/ldap.py index 9c2b0c60..794d8a4d 100644 --- a/nxc/protocols/ldap.py +++ b/nxc/protocols/ldap.py @@ -29,6 +29,7 @@ from impacket.krb5.types import Principal, KerberosException from impacket.ldap import ldap as ldap_impacket from impacket.ldap import ldaptypes from impacket.ldap import ldapasn1 as ldapasn1_impacket +from impacket.ldap.ldapasn1 import AttributeValue from impacket.ldap.ldap import LDAPFilterSyntaxError from impacket.smb import SMB_DIALECT from impacket.smbconnection import SMBConnection, SessionError @@ -1100,41 +1101,46 @@ class ldap(connection): def printTable(items, header): colLen = [] + + # Calculating maximum lenght before parsing CN. for i, col in enumerate(header): - rowMaxLen = max(len(str(row[i])) for row in items) + rowMaxLen = max(len(row[1].split(",")[0].split("CN=")[-1]) for row in items) if i == 1 else max(len(str(row[i])) for row in items) colLen.append(max(rowMaxLen, len(col))) # Create the format string for each row outputFormat = " ".join([f"{{{num}:{width}s}}" for num, width in enumerate(colLen)]) + # Print header self.logger.highlight(outputFormat.format(*header)) self.logger.highlight(" ".join(["-" * itemLen for itemLen in colLen])) # Print rows for row in items: - # Burada DelegationRightsTo'yu düzeltmek için join() ekleyin + # Get first CN value. + if "CN=" in row[1]: + row[1] = row[1].split(",")[0].split("CN=")[-1] + + # Added join for DelegationRightsTo row[3] = ", ".join(str(x) for x in row[3]) if isinstance(row[3], list) else row[3] + self.logger.highlight(outputFormat.format(*row)) - + # Building the search filter search_filter = ("(&(|(UserAccountControl:1.2.840.113556.1.4.803:=16777216)" "(UserAccountControl:1.2.840.113556.1.4.803:=524288)" "(msDS-AllowedToDelegateTo=*)(msDS-AllowedToActOnBehalfOfOtherIdentity=*))" "(!(UserAccountControl:1.2.840.113556.1.4.803:=2))" "(!(UserAccountControl:1.2.840.113556.1.4.803:=8192)))") - + attributes = ["sAMAccountName", "pwdLastSet", "userAccountControl", "objectCategory", "msDS-AllowedToActOnBehalfOfOtherIdentity", "msDS-AllowedToDelegateTo"] resp = self.search(search_filter, attributes, 0) - answers = [] self.logger.debug(f"Total of records returned {len(resp):d}") + resp_parse = parse_result_attributes(resp) - for item in resp: - if not isinstance(item, ldapasn1_impacket.SearchResultEntry): - continue - + for item in resp_parse: mustCommit = False sAMAccountName = "" userAccountControl = 0 @@ -1142,50 +1148,50 @@ class ldap(connection): objectType = "" rightsTo = [] protocolTransition = 0 - + try: - for attribute in item["attributes"]: - if str(attribute["type"]) == "sAMAccountName": - sAMAccountName = str(attribute["vals"][0]) - mustCommit = True - elif str(attribute["type"]) == "userAccountControl": - userAccountControl = str(attribute["vals"][0]) - if int(userAccountControl) & UF_TRUSTED_FOR_DELEGATION: - delegation = "Unconstrained" - rightsTo.append("N/A") - elif int(userAccountControl) & UF_TRUSTED_TO_AUTHENTICATE_FOR_DELEGATION: - delegation = "Constrained w/ Protocol Transition" - protocolTransition = 1 - elif str(attribute["type"]) == "objectCategory": - objectType = str(attribute["vals"][0]).split("=")[1].split(",")[0] - elif str(attribute["type"]) == "msDS-AllowedToDelegateTo": - if protocolTransition == 0: - delegation = "Constrained" - rightsTo = [processAttributeValue(val) for val in attribute["vals"]] + sAMAccountName = item.get("sAMAccountName") + mustCommit = sAMAccountName is not None - # Not an elif as an object could both have RBCD and another type of delegation - if str(attribute["type"]) == "msDS-AllowedToActOnBehalfOfOtherIdentity": - rbcdRights = [] - rbcdObjType = [] - sd = ldaptypes.SR_SECURITY_DESCRIPTOR(data=bytes(attribute["vals"][0])) - search_filter = "(&(|" - for ace in sd["Dacl"].aces: - search_filter += "(objectSid=" + ace["Ace"]["Sid"].formatCanonical() + ")" - search_filter += ")(!(UserAccountControl:1.2.840.113556.1.4.803:=2)))" - delegUserResp = self.search(search_filter, attributes=["sAMAccountName", "objectCategory"], sizeLimit=999) + userAccountControl = int(item.get("userAccountControl", 0)) + objectType = item.get("objectCategory") - for item2 in delegUserResp: - if not isinstance(item2, ldapasn1_impacket.SearchResultEntry): - continue - rbcdRights.append(str(item2["attributes"][0]["vals"][0])) - rbcdObjType.append(str(item2["attributes"][1]["vals"][0]).split("=")[1].split(",")[0]) + if userAccountControl & UF_TRUSTED_FOR_DELEGATION: + delegation = "Unconstrained" + rightsTo.append("N/A") + elif userAccountControl & UF_TRUSTED_TO_AUTHENTICATE_FOR_DELEGATION: + delegation = "Constrained w/ Protocol Transition" + protocolTransition = 1 - if mustCommit: - if int(userAccountControl) & UF_ACCOUNTDISABLE: - self.logger.debug(f"Bypassing disabled account {sAMAccountName}") - else: - for rights, objType in zip(rbcdRights, rbcdObjType): - answers.append([rights, objType, "Resource-Based Constrained", sAMAccountName]) + if item.get("msDS-AllowedToDelegateTo") is not None: + if protocolTransition == 0: + delegation = "Constrained" + rightsTo = item.get("msDS-AllowedToDelegateTo") + + # Not an elif as an object could both have RBCD and another type of delegation + if item.get("msDS-AllowedToActOnBehalfOfOtherIdentity") is not None: + databyte = AttributeValue(item.get("msDS-AllowedToActOnBehalfOfOtherIdentity")) # STR to impacket.ldap.ldapasn1.AttributeValue + rbcdRights = [] + rbcdObjType = [] + sd = ldaptypes.SR_SECURITY_DESCRIPTOR(data=bytes(databyte)) + search_filter = "(&(|" + for ace in sd["Dacl"].aces: + search_filter += "(objectSid=" + ace["Ace"]["Sid"].formatCanonical() + ")" + search_filter += ")(!(UserAccountControl:1.2.840.113556.1.4.803:=2)))" + delegUserResp = self.search(search_filter, attributes=["sAMAccountName", "objectCategory"], sizeLimit=999) + + for item2 in delegUserResp: + if not isinstance(item2, ldapasn1_impacket.SearchResultEntry): + continue + rbcdRights.append(str(item2["attributes"][0]["vals"][0])) + rbcdObjType.append(str(item2["attributes"][1]["vals"][0]).split("=")[1].split(",")[0]) + + if mustCommit: + if int(userAccountControl) & UF_ACCOUNTDISABLE: + self.logger.debug(f"Bypassing disabled account {sAMAccountName}") + else: + for rights, objType in zip(rbcdRights, rbcdObjType): + answers.append([rights, objType, "Resource-Based Constrained", sAMAccountName]) if delegation in ["Unconstrained", "Constrained", "Constrained w/ Protocol Transition"] and mustCommit: if int(userAccountControl) & UF_ACCOUNTDISABLE: