diff --git a/nxc/connection.py b/nxc/connection.py index d619ffef..4f02cef9 100755 --- a/nxc/connection.py +++ b/nxc/connection.py @@ -568,6 +568,10 @@ class connection: if self.args.pfx_cert or self.args.pfx_base64 or self.args.pem_cert: self.logger.debug("Trying to authenticate using Certificate pfx") + if self.args.protocol == "ldap" and self.args.schannel: + # Schannel maps the certificate to an account server-side, so the supplied username is ignored + with sem: + return self.plaintext_login(self.domain, "", "") if not self.args.username: self.logger.fail("You must specify a username when using certificate authentication") return False diff --git a/nxc/helpers/pfx.py b/nxc/helpers/pfx.py index a0a2c323..b0329941 100644 --- a/nxc/helpers/pfx.py +++ b/nxc/helpers/pfx.py @@ -28,6 +28,7 @@ import os import secrets +import tempfile import hashlib import datetime import random @@ -70,7 +71,7 @@ from minikerberos.common.ccache import CCACHE from impacket.krb5.ccache import CCache as impacket_CCache -from nxc.paths import NXC_PATH +from nxc.paths import NXC_PATH, TMP_PATH from nxc.helpers.path import sanitize_path_component from nxc.logger import nxc_logger @@ -569,3 +570,23 @@ def pfx_auth(self): self.logger.info("Successfully authenticated using Certificate") return True + + +def certificate_options(args): + """Return the LDAPConnection certificate options for the command line material, plus a temporary file the caller must delete.""" + if args.pfx_cert: + return {"pfxfile": args.pfx_cert, "pfxpass": args.pfx_pass}, None + + if args.pem_cert and args.pem_key: + return {"certfile": args.pem_cert, "keyfile": args.pem_key}, None + + if args.pfx_base64: + with open(args.pfx_base64, "rb") as pfx_base64_file: + pfxdata = base64.b64decode(pfx_base64_file.read()) + # impacket loads the bundle from a path, so the decoded PFX has to hit the disk + handle, pfx_path = tempfile.mkstemp(suffix=".pfx", dir=TMP_PATH) + with os.fdopen(handle, "wb") as pfx_file: + pfx_file.write(pfxdata) + return {"pfxfile": pfx_path, "pfxpass": args.pfx_pass}, pfx_path + + raise ValueError("You must either specify a PFX file + optional password or a combination of Cert PEM file and Private key PEM file") diff --git a/nxc/netexec.py b/nxc/netexec.py index b4d87689..81a53ea3 100644 --- a/nxc/netexec.py +++ b/nxc/netexec.py @@ -104,6 +104,10 @@ def main(): nxc_logger.fail("Password is required, even if a key file is used - if no passphrase for key, use `-p ''`") exit(1) + if args.protocol == "ldap" and args.schannel and not (args.pfx_cert or args.pfx_base64 or args.pem_cert): + nxc_logger.fail("--schannel requires a certificate: --pfx-cert, --pfx-base64, or --pem-cert with --pem-key") + exit(1) + if args.use_kcache and not os.environ.get("KRB5CCNAME"): nxc_logger.error("KRB5CCNAME environment variable is not set") exit(1) diff --git a/nxc/protocols/ldap.py b/nxc/protocols/ldap.py index f9c5ec97..412eed94 100644 --- a/nxc/protocols/ldap.py +++ b/nxc/protocols/ldap.py @@ -41,6 +41,7 @@ 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.pfx import certificate_options 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 @@ -65,6 +66,9 @@ ldap_error_status = { "KDC_ERR_PREAUTH_FAILED": "KDC_ERR_PREAUTH_FAILED", } +# LDAP authenticationChoice sent to the DC, as displayed to the operator +auth_methods = {"sasl": "NTLM", "simple": "SIMPLE bind", "external": "Schannel"} + class ldap(connection): def __init__(self, args, db, host): @@ -133,18 +137,12 @@ class ldap(connection): 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 "" + try: + # impacket returns the raw RFC 4532 identity, "u:DOMAIN\user" on a DC + return self.ldap_connection.whoami().split("\\")[-1] + except ldap_impacket.LDAPSessionError as e: + self.logger.debug(f"LDAP whoami request failed: {e}") + return "" def check_ldap_signing(self): self.signing_required = None @@ -343,7 +341,7 @@ class ldap(connection): 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.logger.info(f"Connecting to {ldap_url} - {self.baseDN} - {self.host} using Kerberos") 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 == "": @@ -402,7 +400,7 @@ class ldap(connection): 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.logger.info(f"Connecting to {ldaps_url} - {self.baseDN} - {self.host} using Kerberos") 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 == "": @@ -460,21 +458,42 @@ class ldap(connection): hash_asreproast.write(f"{hash_tgt}\n") return False + cert_options = {} + cert_tmp_file = None + if self.args.schannel: + try: + cert_options, cert_tmp_file = certificate_options(self.args) + except (OSError, ValueError) as e: + self.logger.fail(f"Failed to load the certificate material: {e}") + 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) + authentication_choice = "external" if self.args.schannel else self.auth_choice + self.logger.info(f"Connecting to {ldap_url} - {self.baseDN} - {self.host} using {auth_methods[authentication_choice]}") + 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, **cert_options) + self.ldap_connection.login(self.username, self.password, self.domain, self.lmhash, self.nthash, authenticationChoice=authentication_choice) - # Prepare success credential text - self.logger.success(f"{domain}\\{self.username}:{process_secret(self.password)} {self.mark_pwned()}") + if self.args.schannel: + # The certificate is mapped to an account server-side, so whoami is the only way to know who we are + netbios_domain, _, self.username = self.ldap_connection.whoami().removeprefix("u:").partition("\\") + # The mapped account can belong to another domain of the forest, and whoami only returns its NetBIOS domain name + cross_ref = parse_result_attributes(self.search(f"(&(objectClass=crossRef)(nETBIOSName={netbios_domain}))", ["dnsRoot"], baseDN=f"CN=Partitions,{self.configuration_context}")) + self.domain = cross_ref[0]["dnsRoot"] if cross_ref else netbios_domain + + self.check_if_admin() + + # A certificate bind has no secret to store, the account comes from the certificate mapping + if not self.args.schannel: + self.logger.debug(f"Adding credential: {domain}/{self.username}:{self.password}") + self.db.add_credential("plaintext", domain, self.username, self.password) + + used_certificate = " from certificate" if self.args.schannel else f":{process_secret(self.password)}" + self.logger.success(f"{self.domain}\\{self.username}{used_certificate} {self.mark_pwned()}") if self.username != "": add_user_bh(self.username, self.domain, self.logger, self.config) @@ -482,6 +501,13 @@ class ldap(connection): add_user_bh(f"{self.hostname}$", domain, self.logger, self.config) return True except ldap_impacket.LDAPSessionError as e: + if self.args.schannel: + # The error mapping below reads NTLM status codes, which a certificate bind never returns + if "did not receive any credentials via TLS" in str(e) or "connection is anonymous" in str(e): + self.logger.fail("Schannel authentication failed: the DC did not map the certificate to an account") + else: + self.logger.fail(f"Schannel authentication failed: {e}") + return False 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: @@ -495,7 +521,7 @@ class ldap(connection): 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.logger.info(f"Connecting to {ldaps_url} - {self.baseDN} - {self.host} using {auth_methods[self.auth_choice]}") 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() @@ -526,6 +552,9 @@ class ldap(connection): 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 + finally: + if cert_tmp_file and os.path.exists(cert_tmp_file): + os.remove(cert_tmp_file) def hash_login(self, domain, username, ntlm_hash): self.logger.extra["protocol"] = "LDAP" @@ -562,7 +591,7 @@ class ldap(connection): 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.logger.info(f"Connecting to {ldaps_url} - {self.baseDN} - {self.host} using NT hash") 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() @@ -588,7 +617,7 @@ class ldap(connection): 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.logger.info(f"Connecting to {ldaps_url} - {self.baseDN} - {self.host} using NT hash") 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() @@ -631,7 +660,7 @@ class ldap(connection): 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 != "": + if resp and (self.password != "" or self.lmhash != "" or self.nthash != "" or self.aesKey != "" or self.use_kcache or self.args.schannel) and self.username != "": for item in resp_parsed: self.sid_domain = "-".join(item["objectSid"].split("-")[:-1]) diff --git a/nxc/protocols/ldap/proto_args.py b/nxc/protocols/ldap/proto_args.py index aa836396..e9813c63 100644 --- a/nxc/protocols/ldap/proto_args.py +++ b/nxc/protocols/ldap/proto_args.py @@ -10,6 +10,7 @@ def proto_args(parser, parents): ldap_parser.add_argument("--port", type=int, default=389, action=DefaultTrackingAction, help="LDAP port") ldap_parser.add_argument("--ldap-timeout", type=int, default=3, help="LDAP connection timeout") ldap_parser.add_argument("-d", metavar="DOMAIN", dest="domain", type=str, default=None, help="domain to authenticate to") + ldap_parser.add_argument("--schannel", action="store_true", help="Authenticate with the certificate using Schannel instead of PKINIT") egroup = ldap_parser.add_argument_group("Retrieve hash on the remote DC", "Options to get hashes from Kerberos") egroup.add_argument("--asreproast", help="Output AS_REP response to crack with hashcat to file") diff --git a/poetry.lock b/poetry.lock index 5afb0a41..b0dd800f 100644 --- a/poetry.lock +++ b/poetry.lock @@ -1076,7 +1076,7 @@ all = ["mypy (>=1.11.2)", "pytest (>=8.3.2)", "ruff (>=0.6.2)"] [[package]] name = "impacket" -version = "0.14.0.dev0+20261001.122511.1875828d" +version = "0.14.0.dev0+20261015.95810.75938fcb" description = "Network protocols Constructors and Dissectors" optional = false python-versions = "*" @@ -1100,7 +1100,7 @@ six = "*" type = "git" url = "https://github.com/fortra/impacket" reference = "HEAD" -resolved_reference = "1875828d0f2e987cd89c1bcb45f843708981a199" +resolved_reference = "75938fcb0145b476250940d740f71394676288cb" [[package]] name = "iniconfig" diff --git a/tests/data/test_cert.pfx b/tests/data/test_cert.pfx new file mode 100644 index 00000000..55ba7081 Binary files /dev/null and b/tests/data/test_cert.pfx differ diff --git a/tests/e2e_commands.txt b/tests/e2e_commands.txt index 1798a6ea..db173b3e 100644 --- a/tests/e2e_commands.txt +++ b/tests/e2e_commands.txt @@ -221,6 +221,8 @@ netexec wmi TARGET_HOST -u LOGIN_USERNAME -p LOGIN_PASSWORD KERBEROS -M wifi ##### LDAP netexec {DNS} ldap TARGET_HOST -u LOGIN_USERNAME -p LOGIN_PASSWORD KERBEROS netexec ldap TARGET_HOST -u LOGIN_USERNAME -p LOGIN_PASSWORD KERBEROS --simple-bind +netexec ldap TARGET_HOST --pfx-cert TEST_PFX_FILE --schannel +netexec ldap TARGET_HOST --pfx-cert TEST_PFX_FILE --schannel --port 636 netexec ldap TARGET_HOST -u LOGIN_USERNAME -p LOGIN_PASSWORD KERBEROS --users netexec ldap TARGET_HOST -u LOGIN_USERNAME -p LOGIN_PASSWORD KERBEROS --users-export /tmp/userlistOutputFilename.txt netexec ldap TARGET_HOST -u LOGIN_USERNAME -p LOGIN_PASSWORD KERBEROS --groups diff --git a/tests/e2e_tests.py b/tests/e2e_tests.py index da3afc3f..4b3873ea 100644 --- a/tests/e2e_tests.py +++ b/tests/e2e_tests.py @@ -32,6 +32,7 @@ def get_cli_args(): parser.add_argument("--test-password-file", required=False, default=normpath(join(script_dir, "data", "test_passwords.txt")), help="Path to the file containing test passwords") parser.add_argument("--amsi-bypass-file", required=False, default=normpath(join(script_dir, "data", "test_amsi_bypass.txt")), help="Path to the file containing AMSI bypasses") parser.add_argument("--test-normal-file", required=False, default=normpath(join(script_dir, "data", "test_file.txt")), help="Path to file to upload/download") + parser.add_argument("--test-pfx-file", required=False, default=normpath(join(script_dir, "data", "test_cert.pfx")), help="Path to the PFX certificate for certificate authentication") parser.add_argument("--dns-server", action="store", required=False, help="Specify DNS server") return parser.parse_args() @@ -99,6 +100,7 @@ def replace_command(args, line): .replace("TEST_PASSWORD_FILE", args.test_password_file)\ .replace("AMSI_BYPASS_FILE", args.amsi_bypass_file)\ .replace("TEST_NORMAL_FILE", args.test_normal_file)\ + .replace("TEST_PFX_FILE", args.test_pfx_file)\ .replace("{DNS}", dns_server)\ .replace("/tmp", TMP_PATH) if args.poetry: diff --git a/uv.lock b/uv.lock index b4614374..098d111b 100644 --- a/uv.lock +++ b/uv.lock @@ -3,7 +3,7 @@ revision = 3 requires-python = ">=3.11, <4.0" [options] -exclude-newer = "2026-10-13T16:16:27.356501522Z" +exclude-newer = "0001-01-01T00:00:00Z" # This has no effect and is included for backwards compatibility when using relative exclude-newer values. exclude-newer-span = "P1D" [options.exclude-newer-package] @@ -878,8 +878,8 @@ wheels = [ [[package]] name = "impacket" -version = "0.14.0.dev0+20261001.122511.1875828d" -source = { git = "https://github.com/fortra/impacket#1875828d0f2e987cd89c1bcb45f843708981a199" } +version = "0.14.0.dev0+20261015.95810.75938fcb" +source = { git = "https://github.com/fortra/impacket#75938fcb0145b476250940d740f71394676288cb" } dependencies = [ { name = "charset-normalizer" }, { name = "flask" },