Skip to content

Fix custom LDAP port specification and improve scheme handling #1203

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
emilyastranova wants to merge 2 commits into
base: main
Choose a base branch
from
+54 −23
Open

Fix custom LDAP port specification and improve scheme handling #1203

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
16b5b22
Fix custom port specification
emilyastranova Apr 15, 2026
a7987d6
Changed to scheme arg
emilyastranova Apr 15, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
76 changes: 53 additions & 23 deletions nxc/protocols/ldap.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -32,6 +32,26 @@
from impacket.krb5.ccache import CCache
from impacket.krb5.types import Principal, KerberosException
from impacket.ldap import ldap as ldap_impacket
from urllib.parse import urlparse

# Monkey patch LDAPConnection to support custom ports in the URL
_original_ldap_init = ldap_impacket.LDAPConnection.__init__

def _patched_ldap_init(self, url, baseDN='', dstIp=None, signing=True):
parsed = urlparse(url)
self._custom_port = parsed.port
_original_ldap_init(self, url, baseDN=baseDN, dstIp=dstIp, signing=signing)

ldap_impacket.LDAPConnection.__init__ = _patched_ldap_init

def _dstPort_getter(self):
return self._custom_port if getattr(self, '_custom_port', None) else self._original_dstPort

def _dstPort_setter(self, value):
self._original_dstPort = value

ldap_impacket.LDAPConnection._dstPort = property(_dstPort_getter, _dstPort_setter)

from impacket.ldap import ldaptypes
from impacket.ldap import ldapasn1 as ldapasn1_impacket
from impacket.ldap.ldap import LDAPFilterSyntaxError
Expand Down Expand Up @@ -92,10 +112,19 @@ def __init__(self, args, db, host):

connection.__init__(self, args, db, host)

def proto_flow(self):
self.protocol = getattr(self.args, "scheme", "ldaps")
if not getattr(self.args, "port_explicitly_set", False):
if self.protocol == "ldaps":
self.port = 636
else:
self.port = 389
super().proto_flow()

def proto_logger(self):
self.logger = NXCAdapter(
extra={
"protocol": "LDAP",
"protocol": self.protocol.upper(),
"host": self.host,
"port": self.port,
"hostname": self.hostname,
Expand All @@ -104,8 +133,8 @@ def proto_logger(self):

def create_conn_obj(self):
try:
proto = "ldaps" if self.port == 636 else "ldap"
ldap_url = f"{proto}://{self.host}"
proto = self.protocol
ldap_url = f"{proto}://{self.host}:{self.port}"
self.logger.info(f"Connecting to {ldap_url} with no baseDN")

self.ldap_connection = ldap_impacket.LDAPConnection(ldap_url, dstIp=self.host)
Expand Down Expand Up @@ -148,7 +177,7 @@ def get_ldap_username(self):

def check_ldap_signing(self):
self.signing_required = False
ldap_url = f"ldap://{self.target}"
ldap_url = f"ldap://{self.target}:{self.port}"
try:
ldap_connection = ldap_impacket.LDAPConnection(url=ldap_url, baseDN=self.baseDN, dstIp=self.host, signing=False)
ldap_connection.login(domain=self.domain)
Expand All @@ -162,7 +191,8 @@ def check_ldap_signing(self):

def check_ldaps_cbt(self):
self.cbt_status = "Never"
ldap_url = f"ldaps://{self.target}"
ldaps_port = 636 if self.protocol == "ldap" else self.port
ldap_url = f"ldaps://{self.target}:{ldaps_port}"
try:
ldap_connection = ldap_impacket.LDAPConnection(url=ldap_url, baseDN=self.baseDN, dstIp=self.host)
ldap_connection.channel_binding_value = None
Expand Down Expand Up @@ -283,7 +313,7 @@ def print_host_info(self):
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["protocol"] = self.protocol.upper()
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}")
Expand Down Expand Up @@ -326,10 +356,10 @@ def kerberos_login(self, domain, username, password="", ntlm_hash="", aesKey="",

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.extra["protocol"] = self.protocol.upper()
self.logger.extra["port"] = self.port
proto = self.protocol
ldap_url = f"{proto}://{self.target}:{self.port}"
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)
self.ldap_connection.kerberosLogin(username, password, domain, self.lmhash, self.nthash, aesKey, kdcHost=kdcHost, useCache=useCache)
Expand Down Expand Up @@ -388,7 +418,7 @@ def kerberos_login(self, domain, username, password="", ntlm_hash="", aesKey="",
self.logger.extra["protocol"] = "LDAPS"
self.logger.extra["port"] = "636"
self.port = 636
ldaps_url = f"ldaps://{self.target}"
ldaps_url = f"ldaps://{self.target}:{self.port}"
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)
self.ldap_connection.kerberosLogin(username, password, domain, self.lmhash, self.nthash, aesKey, kdcHost=kdcHost, useCache=useCache)
Expand Down Expand Up @@ -449,10 +479,10 @@ def plaintext_login(self, domain, username, password):

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.extra["protocol"] = self.protocol.upper()
self.logger.extra["port"] = self.port
proto = self.protocol
ldap_url = f"{proto}://{self.target}:{self.port}"
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")
self.ldap_connection.login(self.username, self.password, self.domain, self.lmhash, self.nthash, authenticationChoice=self.auth_choice)
Expand Down Expand Up @@ -481,7 +511,7 @@ def plaintext_login(self, domain, username, password):
self.logger.extra["protocol"] = "LDAPS"
self.logger.extra["port"] = "636"
self.port = 636
ldaps_url = f"ldaps://{self.target}"
ldaps_url = f"ldaps://{self.target}:{self.port}"
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)
self.ldap_connection.login(self.username, self.password, self.domain, self.lmhash, self.nthash, authenticationChoice=self.auth_choice)
Expand Down Expand Up @@ -515,8 +545,8 @@ def plaintext_login(self, domain, username, password):
return False

def hash_login(self, domain, username, ntlm_hash):
self.logger.extra["protocol"] = "LDAP"
self.logger.extra["port"] = "389"
self.logger.extra["protocol"] = self.protocol.upper()
self.logger.extra["port"] = self.port
lmhash = ""
nthash = ""

Expand Down Expand Up @@ -545,10 +575,10 @@ def hash_login(self, domain, username, ntlm_hash):

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.extra["protocol"] = self.protocol.upper()
self.logger.extra["port"] = self.port
proto = self.protocol
ldaps_url = f"{proto}://{self.target}:{self.port}"
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)
self.ldap_connection.login(self.username, self.password, self.domain, self.lmhash, self.nthash)
Expand All @@ -574,7 +604,7 @@ def hash_login(self, domain, username, ntlm_hash):
self.logger.extra["protocol"] = "LDAPS"
self.logger.extra["port"] = "636"
self.port = 636
ldaps_url = f"ldaps://{self.target}"
ldaps_url = f"ldaps://{self.target}:{self.port}"
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)
self.ldap_connection.login(self.username, self.password, self.domain, self.lmhash, self.nthash)
Expand Down
1 change: 1 addition & 0 deletions nxc/protocols/ldap/proto_args.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -8,6 +8,7 @@ def proto_args(parser, parents):
dgroup.add_argument("-H", "--hash", metavar="HASH", dest="hash", nargs="+", default=[], help="NTLM hash(es) or file(s) containing NTLM hashes")
dgroup.add_argument("--simple-bind", action="store_true", help="Use simple bind authentication (no signing/sealing)")
ldap_parser.add_argument("--port", type=int, default=389, action=DefaultTrackingAction, help="LDAP port")
ldap_parser.add_argument("--scheme", type=str, choices=["ldap", "ldaps"], default="ldaps", help="Protocol to use (ldap or ldaps).")
ldap_parser.add_argument("-d", metavar="DOMAIN", dest="domain", type=str, default=None, help="domain to authenticate to")

egroup = ldap_parser.add_argument_group("Retrieve hash on the remote DC", "Options to get hashes from Kerberos")
Expand Down
Loading