Merge pull request #1277 from azoxlpf/feat/add-schannel

add Schannel certificate authentication via --schannel

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

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")

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)

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])
 

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")

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"

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

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:

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" },