From a91761a755ea63ac5141eab201a3de361d50ee4e Mon Sep 17 00:00:00 2001 From: Fox Date: Thu, 6 Mar 2025 14:03:06 -0800 Subject: [PATCH] Improve reliability of ldap-checker module --- nxc/modules/ldap-checker.py | 342 ++++++++++++++++++++---------------- nxc/protocols/ldap.py | 2 +- 2 files changed, 193 insertions(+), 151 deletions(-) diff --git a/nxc/modules/ldap-checker.py b/nxc/modules/ldap-checker.py index 3a13f2cc..59e9550a 100644 --- a/nxc/modules/ldap-checker.py +++ b/nxc/modules/ldap-checker.py @@ -1,6 +1,8 @@ import socket import ssl import asyncio +import hashlib +import random from msldap.connection import MSLDAPClientConnection from msldap.commons.target import MSLDAPTarget @@ -10,19 +12,18 @@ from asyauth.common.credentials.ntlm import NTLMCredential from asyauth.common.credentials.kerberos import KerberosCredential from asysocks.unicomm.common.target import UniTarget, UniProto -import sys +import contextlib class NXCModule: """ - Checks whether LDAP signing and channelbinding are required. + Checks whether LDAP signing and LDAPS channel binding are required and/or enforced. - Module by LuemmelSec (@theluemmel), updated by @zblurx + Module by LuemmelSec (@theluemmel), updated by @zblurx/@Mercury0 Original work thankfully taken from @zyn3rgy's Ldap Relay Scan project: https://github.com/zyn3rgy/LdapRelayScan """ - name = "ldap-checker" - description = "Checks whether LDAP signing and binding are required and / or enforced" + description = "Checks whether LDAP signing and channel binding are required and / or enforced" supported_protocols = ["ldap"] opsec_safe = True multiple_hosts = True @@ -30,122 +31,149 @@ class NXCModule: def options(self, context, module_options): """No options available.""" - def on_login(self, context, connection): - # Conduct a bind to LDAPS and determine if channel - # binding is enforced based on the contents of potential - # errors returned. This can be determined unauthenticated, - # because the error indicating channel binding enforcement - # will be returned regardless of a successful LDAPS bind. - async def run_ldaps_noEPA(target, credential): - ldapsClientConn = MSLDAPClientConnection(target, credential) - _, err = await ldapsClientConn.connect() - - # Required step to try to bind without channel binding - ldapsClientConn.cb_data = None - - if err is not None: - context.log.fail("ERROR while connecting to " + str(connection.domain) + ": " + str(err)) - sys.exit() - - valid, err = await ldapsClientConn.bind() - if "data 80090346" in str(err): - return True # channel binding IS enforced - elif "data 52e" in str(err): - return False # channel binding not enforced - elif err is None: - # LDAPS bind successful - # because channel binding is not enforced - return False - - # Conduct a bind to LDAPS with channel binding supported - # but intentionally miscalculated. In the case that and - # LDAPS bind has without channel binding supported has occurred, - # you can determine whether the policy is set to "never" or - # if it's set to "when supported" based on the potential - # error received from the bind attempt. - async def run_ldaps_withEPA(target, credential): - ldapsClientConn = MSLDAPClientConnection(target, credential) - _, err = await ldapsClientConn.connect() - if err is not None: - context.log.fail("ERROR while connecting to " + str(connection.domain) + ": " + str(err)) - sys.exit() - # forcing a miscalculation of the "Channel Bindings" av pair in Type 3 NTLM message - ldapsClientConn.cb_data = b"\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00" - _, err = await ldapsClientConn.bind() - if "data 80090346" in str(err): - return True - elif "data 52e" in str(err): - return False - elif err is not None: - context.log.fail("ERROR while connecting to " + str(connection.domain) + ": " + str(err)) - elif err is None: - return False - - # Domain Controllers do not have a certificate setup for - # LDAPS on port 636 by default. If this has not been setup, - # the TLS handshake will hang and you will not be able to - # interact with LDAPS. The condition for the certificate - # existing as it should is either an error regarding - # the fact that the certificate is self-signed, or - # no error at all. Any other "successful" edge cases - # not yet accounted for. - def DoesLdapsCompleteHandshake(dcIp): - s = socket.socket(socket.AF_INET, socket.SOCK_STREAM) - s.settimeout(5) - ssl_context = ssl.create_default_context() - ssl_context.check_hostname = False - ssl_sock = ssl_context.wrap_socket( - s, - do_handshake_on_connect=False, - suppress_ragged_eofs=False, - ) - try: - ssl_sock.connect((dcIp, 636)) - ssl_sock.do_handshake() - ssl_sock.close() - return True - except Exception as e: - if "CERTIFICATE_VERIFY_FAILED" in str(e): - ssl_sock.close() - return True - if "handshake operation timed out" in str(e): - ssl_sock.close() - return False - else: - context.log.fail("Unexpected error during LDAPS handshake: " + str(e)) - ssl_sock.close() - return False - - # Conduct and LDAP bind and determine if server signing - # requirements are enforced based on potential errors - # during the bind attempt. - async def run_ldap(target, credential): - try: - ldapsClientConn = MSLDAPClientConnection(target, credential) - ldapsClientConn._disable_signing = True - _, err = await ldapsClientConn.connect() - if err is not None: - context.log.fail(str(err)) - return None - - _, err = await ldapsClientConn.bind() - if err is not None: - errstr = str(err).lower() - if "stronger" in errstr: - return True - # because LDAP server signing requirements ARE enforced - else: - context.log.fail(str(err)) - else: - # LDAPS bind successful - return False - # because LDAP server signing requirements are not enforced - except Exception as e: - context.log.debug(str(e)) + # Conduct a bind to LDAPS and determine if channel + # binding is enforced based on the contents of potential + # errors returned. This can be determined unauthenticated, + # because the error indicating channel binding enforcement + # will be returned regardless of a successful LDAPS bind. + async def run_ldaps_noEPA(self, context, connection, target, credential): + try: + client = MSLDAPClientConnection(target, credential) + _, err = await client.connect() + if err: + context.log.debug(f"Error connecting to {connection.domain}: {err}") return None - - # Run trough all our code blocks to determine LDAP signing and channel binding settings. + client.cb_data = None + _, err = await client.bind() + if err and "data 80090346" in str(err): + return True # -> channel binding IS enforced + elif err and "data 52e" in str(err): + return False # -> channel binding not enforced + elif err is None: + return False # LDAPS bind successful -> channel binding not enforced + else: + context.log.debug(f"Unexpected error during LDAPS bind (noEPA): {err}") + return None + except Exception as e: + context.log.debug(f"Exception in run_ldaps_noEPA: {e}") + return None + finally: + with contextlib.suppress(Exception): + await client.disconnect() + + # Conduct a bind to LDAPS with channel binding supported + # but intentionally miscalculated. In the case that an + # LDAPS bind without channel binding supported has occurred, + # you can determine whether the policy is set to "never" or + # if it's set to "when supported" based on the potential + # error received from the bind attempt. + async def run_ldaps_withEPA(self, context, connection, target, credential): + try: + client = MSLDAPClientConnection(target, credential) + _, err = await client.connect() + if err: + context.log.fail(f"Error connecting to {connection.domain}: {err}") + return None + + try: + context.log.debug("Retrieving TLS certificate hash...") + ssl_context = ssl.create_default_context() + ssl_context.check_hostname = False + ssl_context.verify_mode = ssl.CERT_NONE + + with socket.create_connection((connection.host, 636)) as sock, ssl_context.wrap_socket(sock, server_hostname=connection.host) as ssl_sock: + cert = ssl_sock.getpeercert(binary_form=True) + + if cert: + cert_hash = hashlib.sha256(cert).digest() + context.log.debug(f"Original certificate hash: {cert_hash.hex()}") + pos = random.randint(0, len(cert_hash) - 1) + tampered_bytes = bytearray(cert_hash) + tampered_bytes[pos] = (tampered_bytes[pos] + 1) % 256 + context.log.debug(f"Tampered certificate hash: {bytes(tampered_bytes).hex()}") + context.log.debug(f"Modified byte at position {pos}") + client.cb_data = b"tls-server-end-point:" + bytes(tampered_bytes) + else: + client.cb_data = b"\x00" * 64 + except Exception as e: + context.log.debug(f"Failed to retrieve TLS certificate hash: {e}") + client.cb_data = b"\x00" * 64 + + _, err = await client.bind() + if err and "data 80090346" in str(err): + return True + elif (err and "data 52e" in str(err)) or err is None: + return False + else: + context.log.fail(f"Unexpected error during LDAPS bind (withEPA): {err}") + return None + except Exception as e: + context.log.fail(f"Exception in run_ldaps_withEPA: {e}") + return None + + + # Domain Controllers do not have a certificate setup for + # LDAPS on port 636 by default. If this has not been setup, + # the TLS handshake will hang and you will not be able to + # interact with LDAPS. The condition for the certificate + # existing as it should is either an error regarding + # the fact that the certificate is self-signed, or + # no error at all. Any other "successful" edge cases + # not yet accounted for. + def does_ldaps_complete_handshake(self, context, dc_ip): + s = socket.socket(socket.AF_INET, socket.SOCK_STREAM) + s.settimeout(5) + ssl_context = ssl.create_default_context() + ssl_context.check_hostname = False + ssl_sock = ssl_context.wrap_socket(s, do_handshake_on_connect=False, suppress_ragged_eofs=False) + try: + ssl_sock.connect((dc_ip, 636)) + ssl_sock.do_handshake() + return True + except Exception as e: + if "CERTIFICATE_VERIFY_FAILED" in str(e): + return True + elif "handshake operation timed out" in str(e): + return False + else: + context.log.fail(f"Unexpected error during LDAPS handshake: {e}") + return False + finally: + ssl_sock.close() + + # Conduct an LDAP bind and determine if server signing + # requirements are enforced based on potential errors + # during the bind attempt. + async def run_ldap(self, context, target, credential): + try: + client = MSLDAPClientConnection(target, credential) + client._disable_signing = True # deliberately disable LDAP signing on client connection + _, err = await client.connect() + if err: + context.log.fail(f"Error connecting for LDAP bind: {err}") + return None + + _, err = await client.bind() + if err: + errstr = str(err).lower() + if "stronger" in errstr: + return True + # because LDAP server signing requirements ARE enforced + else: + context.log.fail(f"LDAP bind error: {err}") + return None + else: + # LDAPS bind successful + return False + # because LDAP server signing requirements are not enforced + except Exception as e: + context.log.debug(f"Exception during LDAP bind: {e}") + return None + + # Determine authentication context and proceed to + # enumerate LDAP signing and channel binding settings + def on_login(self, context, connection): stype = asyauthSecret.PASS secret = connection.password if connection.nthash: @@ -154,21 +182,24 @@ class NXCModule: if connection.aesKey: stype = asyauthSecret.AES secret = connection.aesKey - if connection.username == "" and secret == "": - credential = NTLMCredential( - secret=None, - username="Guest", - domain=None, - stype=stype, - ) - context.log.info("No username used, skipping LDAP signing check") + + anon_credential = NTLMCredential( + secret="", + username="", + domain=connection.domain, + stype=asyauthSecret.PASS + ) + + if not connection.username and not secret: + context.log.highlight("No credentials provided, skipping LDAP signing check") + credential = anon_credential else: if not connection.kerberos: credential = NTLMCredential( secret=secret, username=connection.username, domain=connection.domain, - stype=stype, + stype=stype ) else: kerberos_target = UniTarget( @@ -189,29 +220,40 @@ class NXCModule: stype=stype, ) - target = MSLDAPTarget(connection.host, 389, hostname=connection.remoteName, domain=connection.domain, dc_ip=connection.kdcHost) - ldapIsProtected = asyncio.run(run_ldap(target, credential)) - if ldapIsProtected is False: - context.log.highlight("LDAP Signing NOT Enforced!") - elif ldapIsProtected is True: - context.log.fail("LDAP Signing IS Enforced") + ldap_signing_status = None + if connection.username or secret: + target = MSLDAPTarget( + connection.host, 389, + hostname=connection.remoteName, + domain=connection.domain, + dc_ip=connection.kdcHost, + ) + ldap_signing_status = asyncio.run(self.run_ldap(context, target, credential)) + if ldap_signing_status is True: + context.log.highlight("LDAP signing IS enforced") + elif ldap_signing_status is False: + context.log.highlight("LDAP signing NOT enforced") else: - context.log.fail("Connection fail, exiting now") - sys.exit() + context.log.fail("Could not determine LDAP signing requirement.") - if DoesLdapsCompleteHandshake(connection.host) is True: - target = MSLDAPTarget(connection.host, 636, UniProto.CLIENT_SSL_TCP, hostname=connection.remoteName, domain=connection.domain, dc_ip=connection.kdcHost) - ldapsChannelBindingAlwaysCheck = asyncio.run(run_ldaps_noEPA(target, credential)) - target = MSLDAPTarget(connection.host, 636, UniProto.CLIENT_SSL_TCP, hostname=connection.remoteName, domain=connection.domain, dc_ip=connection.kdcHost) - ldapsChannelBindingWhenSupportedCheck = asyncio.run(run_ldaps_withEPA(target, credential)) - if ldapsChannelBindingAlwaysCheck is False and ldapsChannelBindingWhenSupportedCheck is True: - context.log.highlight('LDAPS Channel Binding is set to "When Supported"') - elif ldapsChannelBindingAlwaysCheck is False and ldapsChannelBindingWhenSupportedCheck is False: - context.log.highlight('LDAPS Channel Binding is set to "NEVER"') - elif ldapsChannelBindingAlwaysCheck is True: - context.log.fail('LDAPS Channel Binding is set to "Required"') + if self.does_ldaps_complete_handshake(context, connection.host): + target = MSLDAPTarget( + connection.host, 636, + UniProto.CLIENT_SSL_TCP, + hostname=connection.remoteName, + domain=connection.domain, + dc_ip=connection.kdcHost, + ) + ldaps_noEPA = asyncio.run(self.run_ldaps_noEPA(context, connection, target, anon_credential)) + ldaps_withEPA = asyncio.run(self.run_ldaps_withEPA(context, connection, target, anon_credential)) + + if ldaps_noEPA is False and ldaps_withEPA is True: + context.log.highlight("LDAPS channel binding is set to: When Supported") + elif ldaps_noEPA is False and ldaps_withEPA is False: + context.log.highlight("LDAPS channel binding is set to: Never") + elif ldaps_noEPA is True: + context.log.highlight("LDAPS channel binding is set to: Required") else: - context.log.fail("\nSomething went wrong...") - sys.exit() + context.log.fail("Could not determine LDAPS channel binding settings") else: - context.log.fail(connection.domain + " - cannot complete TLS handshake, cert likely not configured") + context.log.fail(f"{connection.domain} - TLS handshake failed; certificate likely not configured") \ No newline at end of file diff --git a/nxc/protocols/ldap.py b/nxc/protocols/ldap.py index 33960d62..226abfda 100644 --- a/nxc/protocols/ldap.py +++ b/nxc/protocols/ldap.py @@ -573,7 +573,7 @@ class ldap(connection): attributes = ["objectSid"] resp = self.search(search_filter, attributes, sizeLimit=0) answers = [] - if resp and (self.password != "" or self.lmhash != "" or self.nthash != "") and self.username != "": + if resp and (self.password != "" or self.lmhash != "" or self.nthash != "" or self.aesKey != "") and self.username != "": for attribute in resp[0][1]: if str(attribute["type"]) == "objectSid": sid = self.sid_to_str(attribute["vals"][0])