add Schannel certificate authentication via --schannel

nxc/cli.py

@@ -107,6 +107,7 @@ def gen_cli_args():
     certificate_group.add_argument("--pfx-pass", metavar="PFXPASS", help="Password of the pfx certificate")
     certificate_group.add_argument("--pem-cert", metavar="PEMCERT", help="Use certificate authentication from PEM file")
     certificate_group.add_argument("--pem-key", metavar="PEMKEY", help="Private key for the PEM format")
+    certificate_group.add_argument("--schannel", action="store_true", help="Authenticate with the certificate using Schannel instead of PKINIT (LDAP protocol only)")
 
     p_loader = ProtocolLoader()
     protocols = p_loader.get_protocols()

nxc/connection.py

@@ -567,6 +567,12 @@ def login(self):
             if not self.args.username:
                 self.logger.fail("You must specify a username when using certificate authentication")
                 return False
+            if self.args.schannel:
+                if self.args.protocol != "ldap":
+                    self.logger.fail("Schannel authentication is only supported for the LDAP protocol")
+                    return False
+                with sem:
+                    return self.plaintext_login(self.domain, self.args.username[0], "")
             with sem:
                 return pfx_auth(self)
 

nxc/helpers/pfx.py

@@ -65,9 +65,13 @@
 from minikerberos.common.target import KerberosTarget
 from minikerberos.common.ccache import CCACHE
 
+from cryptography.hazmat.primitives import serialization
+from cryptography.hazmat.primitives.serialization import pkcs12
+from cryptography.x509 import load_pem_x509_certificate
+
 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.logger import nxc_logger
 
 
@@ -533,3 +537,40 @@ def pfx_auth(self):
 
     self.logger.info("Successfully authenticated using Certificate")
     return True
+
+
+def pfx_to_pem_files(self):
+    """Convert the provided certificate material (PFX or PEM) into PEM cert and key files in TMP_PATH."""
+    if self.args.pfx_cert or self.args.pfx_base64:
+        if self.args.pfx_base64:
+            with open(self.args.pfx_base64, "rb") as f:
+                pfx_data = base64.b64decode(f.read())
+        else:
+            with open(self.args.pfx_cert, "rb") as f:
+                pfx_data = f.read()
+        pfx_pass = self.args.pfx_pass.encode() if self.args.pfx_pass else None
+        key, cert, _ = pkcs12.load_key_and_certificates(pfx_data, pfx_pass)
+    elif self.args.pem_cert and self.args.pem_key:
+        with open(self.args.pem_cert, "rb") as f:
+            cert = load_pem_x509_certificate(f.read())
+        with open(self.args.pem_key, "rb") as f:
+            key = serialization.load_pem_private_key(f.read(), password=None)
+    else:
+        self.logger.fail("You must either specify a PFX file + optional password or a combination of Cert PEM file and Private key PEM file")
+        return None, None
+
+    basename = f"{self.hostname}_{self.host}_{datetime.datetime.now().strftime('%Y-%m-%d_%H%M%S')}"
+    cert_path = os.path.normpath(os.path.expanduser(f"{TMP_PATH}/{basename}_cert.pem"))
+    key_path = os.path.normpath(os.path.expanduser(f"{TMP_PATH}/{basename}_key.pem"))
+
+    with open(cert_path, "wb") as cert_file:
+        cert_file.write(cert.public_bytes(serialization.Encoding.PEM))
+
+    with open(key_path, "wb") as key_file:
+        key_file.write(key.private_bytes(
+            encoding=serialization.Encoding.PEM,
+            format=serialization.PrivateFormat.TraditionalOpenSSL,
+            encryption_algorithm=serialization.NoEncryption(),
+        ))
+
+    return cert_path, key_path

nxc/protocols/ldap.py

@@ -41,6 +41,7 @@
 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 pfx_to_pem_files
 from nxc.helpers.misc import get_bloodhound_info, convert, d2b, parse_argument
 from nxc.logger import NXCAdapter
 from nxc.protocols.ldap.bloodhound import BloodHound, resolve_collection_methods
@@ -447,21 +448,43 @@ def plaintext_login(self, domain, username, password):
                     hash_asreproast.write(f"{hash_tgt}\n")
             return False
 
+        cert_file = key_file = None
+        if self.args.schannel:
+            cert_file, key_file = pfx_to_pem_files(self)
+            if not cert_file:
+                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")
-            self.ldap_connection.login(self.username, self.password, self.domain, self.lmhash, self.nthash, authenticationChoice=self.auth_choice)
+            if self.args.schannel:
+                # Schannel needs TLS: over LDAPS the certificate is sent during the handshake, over LDAP it is sent during StartTLS
+                self.logger.info(f"Connecting to {ldap_url} using Schannel")
+                self.ldap_connection = ldap_impacket.LDAPConnection(url=ldap_url, baseDN=self.baseDN, dstIp=self.host, certfile=cert_file, keyfile=key_file)
+                self.ldap_connection.login(authenticationChoice="external")
+
+                mapped_user = self.get_ldap_username()
+                if mapped_user:
+                    self.username = mapped_user
+            else:
+                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)
+
             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.args.schannel:
+                self.logger.debug(f"Adding credential: {self.domain}/{self.username} from certificate")
+                self.db.add_credential("certificate", self.domain, self.username, "")
+                self.logger.success(f"{self.domain}\\{self.username} from certificate {self.mark_pwned()}")
+            else:
+                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)
@@ -513,6 +536,10 @@ def plaintext_login(self, domain, username, password):
         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:
+            for tmp_file in (cert_file, key_file):
+                if tmp_file and os.path.exists(tmp_file):
+                    os.remove(tmp_file)
 
     def hash_login(self, domain, username, ntlm_hash):
         self.logger.extra["protocol"] = "LDAP"
@@ -618,7 +645,7 @@ def check_if_admin(self):
         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])
 

pyproject.toml

@@ -23,6 +23,7 @@ dependencies = [
     "beautifulsoup4>=4.11,<5",
     "bloodhound-ce>=1.8.0",
     "certihound>=0.1.1",
+    "cryptography>=42.0.8",
     "dploot>=3.2.2",
     "dsinternals>=1.2.4",
     "jwt>=1.3.1",